You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluo.apache.org by ct...@apache.org on 2017/11/10 14:48:33 UTC
[fluo] branch master updated (4207df4 -> eb0889b)
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/fluo.git.
from 4207df4 fixed #954 Appended FLUO_JAVA_OPTS to JAVA_OPTS in fluo-env.sh (#966)
new 22d17f9 Minor code cleanup and remove compiler warnings
new eb0889b Use NewTableConfiguration when creating tables
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.travis.yml | 4 +-
.../fluo/accumulo/iterators/ColumnBuffer.java | 26 ++++---
.../iterators/GarbageCollectionIterator.java | 11 ++-
.../apache/fluo/accumulo/util/AccumuloProps.java | 7 +-
.../apache/fluo/accumulo/util/ColumnConstants.java | 6 +-
.../apache/fluo/accumulo/iterators/TestData.java | 2 +-
.../fluo/api/client/AbstractSnapshotBase.java | 2 +-
.../api/client/AbstractTransactionBaseTest.java | 2 +-
.../fluo/api/config/FluoConfigurationTest.java | 28 +++++---
.../fluo/cluster/runnable/OracleRunnable.java | 7 +-
.../fluo/cluster/runnable/WorkerRunnable.java | 7 +-
.../org/apache/fluo/cluster/runner/AppRunner.java | 8 +--
.../apache/fluo/cluster/runner/YarnAppRunner.java | 51 ++++++++------
.../org/apache/fluo/cluster/yarn/FluoTwillApp.java | 56 +++++++++------
.../org/apache/fluo/core/client/FluoAdminImpl.java | 82 +++++++++++++++-------
.../org/apache/fluo/core/impl/LockResolver.java | 7 +-
.../fluo/core/observer/v1/ObserverStoreV1.java | 57 ++++++++-------
.../apache/fluo/core/observer/v1/ObserversV1.java | 24 ++++---
.../fluo/core/observer/v2/ObserverStoreV2.java | 4 +-
.../core/worker/finder/hash/PartitionInfo.java | 9 ++-
.../fluo/core/worker/finder/hash/RangeSet.java | 9 ++-
.../fluo/core/observer/ObserverStoreTest.java | 7 +-
.../fluo/integration/accumulo/TimeskippingIT.java | 3 +-
.../fluo/integration/client/FluoAdminImplIT.java | 32 ++++++++-
.../fluo/integration/impl/ObserverConfigIT.java | 27 +++----
.../fluo/integration/impl/ReadLockFailureIT.java | 21 ++----
.../org/apache/fluo/integration/impl/WorkerIT.java | 3 +-
.../apache/fluo/integration/impl/ZKSecretIT.java | 8 +--
.../org/apache/fluo/integration/log/LogIT.java | 1 -
pom.xml | 2 +-
30 files changed, 299 insertions(+), 214 deletions(-)
--
To stop receiving notification emails like this one, please contact
['"commits@fluo.apache.org" <co...@fluo.apache.org>'].
[fluo] 02/02: Use NewTableConfiguration when creating tables
Posted by ct...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/fluo.git
commit eb0889b5609292370b4569c576d2aee87c86f04b
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Fri Nov 10 09:13:04 2017 -0500
Use NewTableConfiguration when creating tables
Use Accumulo's NewTableConfiguration when creating new application
tables. This way, the initial table configuration, including locality
group settings and compaction iterators, are already set on the table at
the moment it is created, eliminating the possibility of race conditions
related to setting configuration after the table is created.
This change requires at least Accumulo 1.7.0.
Travis-CI configuration was updated to test against the latest version
of Accumulo (1.8.1) with the corresponding Thrift version (0.9.3)
Include test to verify locality group serialization can be deserialized
by Accumulo's public API, and adds some constants to refer to Accumulo
properties and column family and locality group names.
---
.travis.yml | 4 +-
.../apache/fluo/accumulo/util/AccumuloProps.java | 7 ++-
.../apache/fluo/accumulo/util/ColumnConstants.java | 6 +-
.../org/apache/fluo/core/client/FluoAdminImpl.java | 72 +++++++++++++++-------
.../fluo/integration/accumulo/TimeskippingIT.java | 3 +-
.../fluo/integration/client/FluoAdminImplIT.java | 32 +++++++++-
pom.xml | 2 +-
7 files changed, 95 insertions(+), 31 deletions(-)
diff --git a/.travis.yml b/.travis.yml
index b44275b..7f0dcf0 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,8 +12,8 @@
# the License.
language: java
jdk:
- - oraclejdk8
-script: mvn verify javadoc:jar
+ - openjdk8
+script: mvn clean verify javadoc:jar -Daccumulo.version=1.8.1 -Dthrift.version=0.9.3
notifications:
irc:
channels:
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/AccumuloProps.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/AccumuloProps.java
index e1966f0..c67b782 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/AccumuloProps.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/AccumuloProps.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -21,5 +21,8 @@ public class AccumuloProps {
public static final String TABLE_CLASSPATH = "table.classpath.context";
public static final String TABLE_BLOCKCACHE_ENABLED = "table.cache.block.enable";
public static final String TABLE_FORMATTER_CLASS = "table.formatter";
+ public static final String TABLE_GROUP_PREFIX = "table.group.";
+ public static final String TABLE_GROUPS_ENABLED = "table.groups.enabled";
+ public static final Object TABLE_ITERATOR_PREFIX = "table.iterator.";
}
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
index c38b9e6..7063adc 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/util/ColumnConstants.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -32,6 +32,8 @@ public class ColumnConstants {
public static final long DATA_PREFIX = 0xa000000000000000L;
public static final long TIMESTAMP_MASK = 0x1fffffffffffffffL;
public static final Bytes NOTIFY_CF = Bytes.of("ntfy");
+ public static final String NOTIFY_LOCALITY_GROUP_NAME = "notify";
+ public static final Bytes GC_CF = Bytes.of("gc");
private ColumnConstants() {}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
index fc2701f..158b7d1 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
@@ -27,8 +27,8 @@ import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Properties;
-import java.util.Set;
import java.util.UUID;
import java.util.regex.Pattern;
@@ -36,7 +36,9 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.iterators.IteratorUtil;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.curator.framework.CuratorFramework;
import org.apache.fluo.accumulo.iterators.GarbageCollectionIterator;
import org.apache.fluo.accumulo.iterators.NotificationIterator;
@@ -47,6 +49,7 @@ import org.apache.fluo.accumulo.util.ZookeeperUtil;
import org.apache.fluo.api.client.FluoAdmin;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Bytes;
import org.apache.fluo.api.exceptions.FluoException;
import org.apache.fluo.core.impl.FluoConfigurationImpl;
import org.apache.fluo.core.observer.ObserverUtil;
@@ -56,7 +59,6 @@ import org.apache.fluo.core.worker.finder.hash.PartitionManager;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.Text;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.slf4j.Logger;
@@ -77,7 +79,6 @@ public class FluoAdminImpl implements FluoAdmin {
public FluoAdminImpl(FluoConfiguration config) {
this.config = config;
-
appRootDir = ZookeeperUtil.parseRoot(config.getAppZookeepers());
rootCurator = CuratorUtil.newRootFluoCurator(config);
rootCurator.start();
@@ -145,7 +146,8 @@ public class FluoAdminImpl implements FluoAdmin {
}
try {
- initialize(conn);
+ initializeApplicationInZooKeeper(conn);
+ Map<String, String> ntcProps = initializeApplicationTableProps();
String accumuloJars;
if (!config.getAccumuloJars().trim().isEmpty()) {
@@ -167,8 +169,7 @@ public class FluoAdminImpl implements FluoAdmin {
String contextName = "fluo-" + config.getApplicationName();
conn.instanceOperations().setProperty(
AccumuloProps.VFS_CONTEXT_CLASSPATH_PROPERTY + contextName, accumuloClasspath);
- conn.tableOperations().setProperty(config.getAccumuloTable(), AccumuloProps.TABLE_CLASSPATH,
- contextName);
+ ntcProps.put(AccumuloProps.TABLE_CLASSPATH, contextName);
}
if (config.getObserverJarsUrl().isEmpty() && !config.getObserverInitDir().trim().isEmpty()) {
@@ -176,8 +177,11 @@ public class FluoAdminImpl implements FluoAdmin {
config.setObserverJarsUrl(observerUrl);
}
- conn.tableOperations().setProperty(config.getAccumuloTable(),
- AccumuloProps.TABLE_BLOCKCACHE_ENABLED, "true");
+ ntcProps.put(AccumuloProps.TABLE_BLOCKCACHE_ENABLED, "true");
+
+ NewTableConfiguration ntc = new NewTableConfiguration().withoutDefaultIterators();
+ ntc.setProperties(ntcProps);
+ conn.tableOperations().create(config.getAccumuloTable(), ntc);
updateSharedConfig();
} catch (NodeExistsException nee) {
@@ -190,7 +194,7 @@ public class FluoAdminImpl implements FluoAdmin {
}
}
- private void initialize(Connector conn) throws Exception {
+ private void initializeApplicationInZooKeeper(Connector conn) throws Exception {
final String accumuloInstanceName = conn.getInstance().getInstanceName();
final String accumuloInstanceID = conn.getInstance().getInstanceID();
@@ -221,23 +225,49 @@ public class FluoAdminImpl implements FluoAdmin {
CuratorUtil.NodeExistsPolicy.FAIL);
CuratorUtil.putData(curator, ZookeeperPath.ORACLE_GC_TIMESTAMP, new byte[] {'0'},
CuratorUtil.NodeExistsPolicy.FAIL);
+ }
- conn.tableOperations().create(config.getAccumuloTable(), false);
- Map<String, Set<Text>> groups = new HashMap<>();
- groups.put("notify", Collections.singleton(ByteUtil.toText(ColumnConstants.NOTIFY_CF)));
- conn.tableOperations().setLocalityGroups(config.getAccumuloTable(), groups);
-
- IteratorSetting gcIter = new IteratorSetting(10, "gc", GarbageCollectionIterator.class);
- GarbageCollectionIterator.setZookeepers(gcIter, config.getAppZookeepers());
+ private String encodeColumnFamily(Bytes cf) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < cf.length(); i++) {
+ int c = 0xff & cf.byteAt(i);
+ if (c == '\\') {
+ sb.append("\\\\");
+ } else if (c >= 32 && c <= 126 && c != ',') {
+ sb.append((char) c);
+ } else {
+ sb.append("\\x").append(String.format("%02X", c));
+ }
+ }
+ return sb.toString();
+ }
- conn.tableOperations().attachIterator(config.getAccumuloTable(), gcIter,
- EnumSet.of(IteratorUtil.IteratorScope.majc, IteratorUtil.IteratorScope.minc));
+ private Map<String, String> initializeApplicationTableProps() {
+ Map<String, String> ntcProps = new HashMap<>();
+ ntcProps.put(AccumuloProps.TABLE_GROUP_PREFIX + ColumnConstants.NOTIFY_LOCALITY_GROUP_NAME,
+ encodeColumnFamily(ColumnConstants.NOTIFY_CF));
+ ntcProps.put(AccumuloProps.TABLE_GROUPS_ENABLED, ColumnConstants.NOTIFY_LOCALITY_GROUP_NAME);
+ IteratorSetting gcIter =
+ new IteratorSetting(10, ColumnConstants.GC_CF.toString(), GarbageCollectionIterator.class);
+ GarbageCollectionIterator.setZookeepers(gcIter, config.getAppZookeepers());
// the order relative to gc iter should not matter
- IteratorSetting ntfyIter = new IteratorSetting(11, "ntfy", NotificationIterator.class);
+ IteratorSetting ntfyIter =
+ new IteratorSetting(11, ColumnConstants.NOTIFY_CF.toString(), NotificationIterator.class);
+
+ for (IteratorSetting setting : new IteratorSetting[] {gcIter, ntfyIter}) {
+ for (IteratorScope scope : EnumSet.of(IteratorUtil.IteratorScope.majc,
+ IteratorUtil.IteratorScope.minc)) {
+ String root = String.format("%s%s.%s", AccumuloProps.TABLE_ITERATOR_PREFIX,
+ scope.name().toLowerCase(), setting.getName());
+ for (Entry<String, String> prop : setting.getOptions().entrySet()) {
+ ntcProps.put(root + ".opt." + prop.getKey(), prop.getValue());
+ }
+ ntcProps.put(root, setting.getPriority() + "," + setting.getIteratorClass());
+ }
+ }
- conn.tableOperations().attachIterator(config.getAccumuloTable(), ntfyIter,
- EnumSet.of(IteratorUtil.IteratorScope.majc, IteratorUtil.IteratorScope.minc));
+ return ntcProps;
}
@Override
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java
index eaecc29..9aeca79 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java
@@ -20,6 +20,7 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.fluo.integration.ITBase;
@@ -39,7 +40,7 @@ public class TimeskippingIT extends ITBase {
@Test
public void testTimestampSkippingIterPerformance() throws Exception {
- conn.tableOperations().create("ttsi", false);
+ conn.tableOperations().create("ttsi", new NewTableConfiguration().withoutDefaultIterators());
BatchWriter bw = conn.createBatchWriter("ttsi", new BatchWriterConfig());
Mutation m = new Mutation("r1");
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/client/FluoAdminImplIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/client/FluoAdminImplIT.java
index 4a6a5cf..02b8edc 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/client/FluoAdminImplIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/client/FluoAdminImplIT.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -15,7 +15,16 @@
package org.apache.fluo.integration.client;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.fluo.accumulo.util.ColumnConstants;
import org.apache.fluo.accumulo.util.ZookeeperUtil;
import org.apache.fluo.api.client.FluoAdmin;
import org.apache.fluo.api.client.FluoAdmin.AlreadyInitializedException;
@@ -26,6 +35,7 @@ import org.apache.fluo.core.client.FluoAdminImpl;
import org.apache.fluo.core.client.FluoClientImpl;
import org.apache.fluo.core.util.CuratorUtil;
import org.apache.fluo.integration.ITBaseImpl;
+import org.apache.hadoop.io.Text;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -92,6 +102,23 @@ public class FluoAdminImplIT extends ITBaseImpl {
InitializationOptions opts =
new InitializationOptions().setClearZookeeper(true).setClearTable(true);
admin.initialize(opts);
+
+ // verify locality groups were set on the table
+ Instance inst =
+ new ZooKeeperInstance(config.getAccumuloInstance(), config.getAccumuloZookeepers());
+ Connector conn = inst.getConnector(config.getAccumuloUser(),
+ new PasswordToken(config.getAccumuloPassword()));
+ Map<String, Set<Text>> localityGroups =
+ conn.tableOperations().getLocalityGroups(config.getAccumuloTable());
+ Assert.assertEquals("Unexpected locality group count.", 1, localityGroups.size());
+ Entry<String, Set<Text>> localityGroup = localityGroups.entrySet().iterator().next();
+ Assert.assertEquals("'notify' locality group not found.",
+ ColumnConstants.NOTIFY_LOCALITY_GROUP_NAME, localityGroup.getKey());
+ Assert.assertEquals("'notify' locality group does not contain exactly 1 column family.", 1,
+ localityGroup.getValue().size());
+ Text colFam = localityGroup.getValue().iterator().next();
+ Assert.assertTrue("'notify' locality group does not contain the correct column family.",
+ ColumnConstants.NOTIFY_CF.contentEquals(colFam.getBytes(), 0, colFam.getLength()));
}
try (FluoClientImpl client = new FluoClientImpl(localConfig)) {
@@ -156,4 +183,5 @@ public class FluoAdminImplIT extends ITBaseImpl {
Assert.assertNotNull(curator.checkExists().forPath(ZookeeperUtil.parseRoot(zk + longPath)));
}
}
+
}
diff --git a/pom.xml b/pom.xml
index 05b3afe..ee9021e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,7 +54,7 @@
<url>https://github.com/apache/fluo/issues</url>
</issueManagement>
<properties>
- <accumulo.version>1.6.5</accumulo.version>
+ <accumulo.version>1.7.3</accumulo.version>
<curator.version>2.7.1</curator.version>
<dropwizard.version>0.8.1</dropwizard.version>
<findbugs.maxRank>11</findbugs.maxRank>
--
To stop receiving notification emails like this one, please contact
"commits@fluo.apache.org" <co...@fluo.apache.org>.
[fluo] 01/02: Minor code cleanup and remove compiler warnings
Posted by ct...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/fluo.git
commit 22d17f92e6f1d81eceaa0e4ff4b6cd17315939fa
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Fri Nov 10 09:12:57 2017 -0500
Minor code cleanup and remove compiler warnings
Remove all compiler warnings from javac on the command-line and in
Eclipse IDE
* Remove unused imports
* Remove redundant generic type specifications
* Add missing Override annotations
* Remove redundant semi-colon
* Apply formatter changes
* Fix broken javadoc
* Work around deprecation warnings by creating temp variables and
suppressing
* Work around https://bugs.openjdk.java.net/browse/JDK-8032211 by using
fully qualified class names instead of imports when a deprecated class
must be used
* Add missing hashCode implementations when equals is overridden, so
there aren't suprising behaviors (throw UnsuppotedOperationException
instead)
* Remove dead/unused private code
---
.../fluo/accumulo/iterators/ColumnBuffer.java | 26 +++++-----
.../iterators/GarbageCollectionIterator.java | 11 ++---
.../apache/fluo/accumulo/iterators/TestData.java | 2 +-
.../fluo/api/client/AbstractSnapshotBase.java | 2 +-
.../api/client/AbstractTransactionBaseTest.java | 2 +-
.../fluo/api/config/FluoConfigurationTest.java | 28 +++++++----
.../fluo/cluster/runnable/OracleRunnable.java | 7 ++-
.../fluo/cluster/runnable/WorkerRunnable.java | 7 ++-
.../org/apache/fluo/cluster/runner/AppRunner.java | 8 +--
.../apache/fluo/cluster/runner/YarnAppRunner.java | 51 ++++++++++---------
.../org/apache/fluo/cluster/yarn/FluoTwillApp.java | 56 ++++++++++++---------
.../org/apache/fluo/core/client/FluoAdminImpl.java | 10 ++--
.../org/apache/fluo/core/impl/LockResolver.java | 7 ++-
.../fluo/core/observer/v1/ObserverStoreV1.java | 57 ++++++++++++----------
.../apache/fluo/core/observer/v1/ObserversV1.java | 24 ++++-----
.../fluo/core/observer/v2/ObserverStoreV2.java | 4 +-
.../core/worker/finder/hash/PartitionInfo.java | 9 +++-
.../fluo/core/worker/finder/hash/RangeSet.java | 9 +++-
.../fluo/core/observer/ObserverStoreTest.java | 7 ++-
.../fluo/integration/impl/ObserverConfigIT.java | 27 +++++-----
.../fluo/integration/impl/ReadLockFailureIT.java | 21 ++------
.../org/apache/fluo/integration/impl/WorkerIT.java | 3 +-
.../apache/fluo/integration/impl/ZKSecretIT.java | 8 ++-
.../org/apache/fluo/integration/log/LogIT.java | 1 -
24 files changed, 204 insertions(+), 183 deletions(-)
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/ColumnBuffer.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/ColumnBuffer.java
index 97be335..fba872b 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/ColumnBuffer.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/ColumnBuffer.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -15,7 +15,6 @@
package org.apache.fluo.accumulo.iterators;
-import java.lang.IllegalArgumentException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.function.LongPredicate;
@@ -25,9 +24,8 @@ import org.apache.accumulo.core.data.PartialKey;
import org.apache.accumulo.core.data.Value;
/**
- * This class buffers Keys that all have the same row+column. Internally
- * it only stores one Key, a list of timestamps and a list of values. At iteration
- * time it materializes each Key+Value.
+ * This class buffers Keys that all have the same row+column. Internally it only stores one Key, a
+ * list of timestamps and a list of values. At iteration time it materializes each Key+Value.
*/
class ColumnBuffer {
@@ -53,11 +51,11 @@ class ColumnBuffer {
}
/**
- * When empty, the first key added sets the row+column. After this all keys
- * added must have the same row+column.
+ * When empty, the first key added sets the row+column. After this all keys added must have the
+ * same row+column.
*
* @param k Key to be added to buffer
- * @param v Value to be added to buffer
+ * @param vByte Value to be added to buffer
*/
public void add(Key k, byte[] vByte) throws IllegalArgumentException {
vByte = Arrays.copyOf(vByte, vByte.length);
@@ -73,8 +71,8 @@ class ColumnBuffer {
}
/**
- * When empty, the first key added sets the row+column. After this all keys
- * added must have the same row+column.
+ * When empty, the first key added sets the row+column. After this all keys added must have the
+ * same row+column.
*
* @param k Key to be added to buffer
* @param v Value to be added to buffer
@@ -84,8 +82,8 @@ class ColumnBuffer {
}
/**
- * Clears the dest ColumnBuffer and inserts all entries in dest where the timestamp passes
- * the timestampTest.
+ * Clears the dest ColumnBuffer and inserts all entries in dest where the timestamp passes the
+ * timestampTest.
*
* @param dest Destination ColumnBuffer
* @param timestampTest Test to determine which timestamps get added to dest
@@ -119,7 +117,7 @@ class ColumnBuffer {
}
/**
- * @param pos Position of the Key that will be retrieved
+ * @param pos Position of the Key that will be retrieved
* @return The key at a given position
*/
public Key getKey(int pos) {
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java
index 1814dfa..d07f59e 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/GarbageCollectionIterator.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -16,12 +16,9 @@
package org.apache.fluo.accumulo.iterators;
import java.io.IOException;
-import java.util.AbstractMap.SimpleImmutableEntry;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
-import java.util.function.LongPredicate;
import com.google.common.annotations.VisibleForTesting;
import org.apache.accumulo.core.client.IteratorSetting;
@@ -87,7 +84,6 @@ public class GarbageCollectionIterator implements SortedKeyValueIterator<Key, Va
}
}
-
@Override
public boolean hasTop() {
return position < keysFiltered.size() || source.hasTop();
@@ -259,7 +255,8 @@ public class GarbageCollectionIterator implements SortedKeyValueIterator<Key, Va
if (isFullMajc) {
if (isDelete) {
if (DelReadLockValue.isRollback(source.getTopValue().get())) {
- // can drop rolled back read lock delete markers on any full majc, do not need to consider gcTimestamp
+ // can drop rolled back read lock delete markers on any full majc, do not need to
+ // consider gcTimestamp
keep = false;
} else {
long rlockCommitTs =
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java
index 262df15..dbdcd1b 100644
--- a/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java
+++ b/modules/accumulo/src/test/java/org/apache/fluo/accumulo/iterators/TestData.java
@@ -83,7 +83,7 @@ public class TestData {
String cq = fields[2];
String ct;
long ts;
- byte[] val = new byte[0];;
+ byte[] val = new byte[0];
if (cf.equals("ntfy")) {
ts = Long.parseLong(fields[3]) << 1;
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java b/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java
index 1ba7f9b..8c0fa4c 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/client/AbstractSnapshotBase.java
@@ -47,7 +47,7 @@ public abstract class AbstractSnapshotBase implements SnapshotBase {
* toString(). Second, the actual Transaction implementation will under some circumstances return
* the Bytes object that was passed in.
*/
- private Map<String, Bytes> s2bCache = new WeakHashMap<String, Bytes>();
+ private Map<String, Bytes> s2bCache = new WeakHashMap<>();
public AbstractSnapshotBase() {}
diff --git a/modules/api/src/test/java/org/apache/fluo/api/client/AbstractTransactionBaseTest.java b/modules/api/src/test/java/org/apache/fluo/api/client/AbstractTransactionBaseTest.java
index fcffe55..7409ca9 100644
--- a/modules/api/src/test/java/org/apache/fluo/api/client/AbstractTransactionBaseTest.java
+++ b/modules/api/src/test/java/org/apache/fluo/api/client/AbstractTransactionBaseTest.java
@@ -69,7 +69,7 @@ public class AbstractTransactionBaseTest {
@Override
public Map<Column, Bytes> get(Bytes row, Set<Column> columns) {
- HashMap<Column, Bytes> ret = new HashMap<Column, Bytes>();
+ HashMap<Column, Bytes> ret = new HashMap<>();
for (Column column : columns) {
RowColumn rc = new RowColumn(row, column);
if (snapshot.containsKey(rc)) {
diff --git a/modules/api/src/test/java/org/apache/fluo/api/config/FluoConfigurationTest.java b/modules/api/src/test/java/org/apache/fluo/api/config/FluoConfigurationTest.java
index 316d0fc..3193f7f 100644
--- a/modules/api/src/test/java/org/apache/fluo/api/config/FluoConfigurationTest.java
+++ b/modules/api/src/test/java/org/apache/fluo/api/config/FluoConfigurationTest.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -49,8 +49,11 @@ public class FluoConfigurationTest {
base.getConnectionRetryTimeout());
Assert.assertEquals(FluoConfiguration.ACCUMULO_ZOOKEEPERS_DEFAULT,
base.getAccumuloZookeepers());
- Assert.assertEquals(FluoConfiguration.ADMIN_ACCUMULO_CLASSPATH_DEFAULT,
- base.getAccumuloClasspath());
+ @SuppressWarnings("deprecation")
+ String tmpFieldName = FluoConfiguration.ADMIN_ACCUMULO_CLASSPATH_DEFAULT;
+ @SuppressWarnings("deprecation")
+ String tmpCP = base.getAccumuloClasspath();
+ Assert.assertEquals(tmpFieldName, tmpCP);
Assert.assertEquals(FluoConfiguration.WORKER_NUM_THREADS_DEFAULT, base.getWorkerThreads());
Assert.assertEquals(FluoConfiguration.TRANSACTION_ROLLBACK_TIME_DEFAULT,
base.getTransactionRollbackTime());
@@ -86,8 +89,9 @@ public class FluoConfigurationTest {
@Test
public void testSetGet() {
FluoConfiguration config = new FluoConfiguration();
- Assert.assertEquals("path1,path2",
- config.setAccumuloClasspath("path1,path2").getAccumuloClasspath());
+ @SuppressWarnings("deprecation")
+ String tmpCP = config.setAccumuloClasspath("path1,path2").getAccumuloClasspath();
+ Assert.assertEquals("path1,path2", tmpCP);
Assert.assertEquals("path1,path2", config.setAccumuloJars("path1,path2").getAccumuloJars());
Assert.assertEquals("instance", config.setAccumuloInstance("instance").getAccumuloInstance());
Assert.assertEquals("pass", config.setAccumuloPassword("pass").getAccumuloPassword());
@@ -184,7 +188,9 @@ public class FluoConfigurationTest {
FluoConfiguration config = new FluoConfiguration(propsFile);
// make sure classpath contains comma. otherwise it was shortened
- Assert.assertTrue(config.getAccumuloClasspath().contains(","));
+ @SuppressWarnings("deprecation")
+ String tmpCP = config.getAccumuloClasspath();
+ Assert.assertTrue(tmpCP.contains(","));
// check for values set in prop file
Assert.assertEquals("localhost/fluo", config.getInstanceZookeepers());
Assert.assertEquals("localhost", config.getAccumuloZookeepers());
@@ -249,7 +255,9 @@ public class FluoConfigurationTest {
FluoConfiguration config = new FluoConfiguration(propsFile);
// make sure classpath contains comma. otherwise it was shortened
- Assert.assertTrue(config.getAccumuloClasspath().contains(","));
+ @SuppressWarnings("deprecation")
+ String tmpCP = config.getAccumuloClasspath();
+ Assert.assertTrue(tmpCP.contains(","));
// check for values set in prop file
Assert.assertEquals("app1", config.getApplicationName());
Assert.assertEquals("localhost/fluo2", config.getInstanceZookeepers());
@@ -258,7 +266,9 @@ public class FluoConfigurationTest {
Assert.assertEquals("user5", config.getAccumuloUser());
Assert.assertEquals("pass6", config.getAccumuloPassword());
Assert.assertEquals("zoo7", config.getAccumuloZookeepers());
- Assert.assertEquals(8, config.getClientRetryTimeout());
+ @SuppressWarnings("deprecation")
+ int tmpTimeout = config.getClientRetryTimeout();
+ Assert.assertEquals(8, tmpTimeout);
Assert.assertEquals(8, config.getConnectionRetryTimeout());
Assert.assertEquals("table9", config.getAccumuloTable());
}
diff --git a/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/OracleRunnable.java b/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/OracleRunnable.java
index 5939710..99d09a0 100644
--- a/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/OracleRunnable.java
+++ b/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/OracleRunnable.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -21,7 +21,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.service.FluoOracle;
-import org.apache.fluo.cluster.util.LogbackUtil;
import org.apache.fluo.core.metrics.MetricNames;
import org.apache.fluo.core.oracle.FluoOracleImpl;
import org.apache.fluo.core.util.UtilWaitThread;
@@ -61,7 +60,7 @@ public class OracleRunnable extends AbstractTwillRunnable {
try {
if (!logDir.equals(STDOUT)) {
- LogbackUtil.init("oracle", configDir, logDir);
+ org.apache.fluo.cluster.util.LogbackUtil.init("oracle", configDir, logDir);
}
} catch (Exception e) {
System.err.println("Exception while starting FluoOracle: " + e.getMessage());
diff --git a/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/WorkerRunnable.java b/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/WorkerRunnable.java
index fa2229b..dbfd191 100644
--- a/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/WorkerRunnable.java
+++ b/modules/cluster/src/main/java/org/apache/fluo/cluster/runnable/WorkerRunnable.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -21,7 +21,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.service.FluoWorker;
-import org.apache.fluo.cluster.util.LogbackUtil;
import org.apache.fluo.core.metrics.MetricNames;
import org.apache.fluo.core.util.UtilWaitThread;
import org.apache.fluo.core.worker.FluoWorkerImpl;
@@ -61,7 +60,7 @@ public class WorkerRunnable extends AbstractTwillRunnable {
try {
if (!logDir.equals(STDOUT)) {
- LogbackUtil.init("worker", configDir, logDir);
+ org.apache.fluo.cluster.util.LogbackUtil.init("worker", configDir, logDir);
}
} catch (Exception e) {
System.err.println("Exception while starting FluoWorker: " + e.getMessage());
diff --git a/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java b/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java
index ca62c69..9de17f4 100644
--- a/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java
+++ b/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -33,7 +33,6 @@ import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.exceptions.FluoException;
-import org.apache.fluo.cluster.util.FluoYarnConfig;
import org.apache.fluo.core.impl.Environment;
import org.apache.fluo.core.impl.Notification;
import org.apache.fluo.core.util.ScanUtil;
@@ -118,7 +117,8 @@ public abstract class AppRunner {
}
try {
- long sleepSec = calculateSleep(ntfyCount, FluoYarnConfig.getWorkerInstances(config));
+ long sleepSec = calculateSleep(ntfyCount,
+ org.apache.fluo.cluster.util.FluoYarnConfig.getWorkerInstances(config));
log.info("{} notifications are still outstanding. Will try again in {} seconds...",
ntfyCount, sleepSec);
Thread.sleep(1000 * sleepSec);
diff --git a/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/YarnAppRunner.java b/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/YarnAppRunner.java
index f5c97c5..be89606 100644
--- a/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/YarnAppRunner.java
+++ b/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/YarnAppRunner.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -30,11 +30,6 @@ import org.apache.curator.framework.CuratorFramework;
import org.apache.fluo.accumulo.util.ZookeeperPath;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.exceptions.FluoException;
-import org.apache.fluo.cluster.runnable.OracleRunnable;
-import org.apache.fluo.cluster.runnable.WorkerRunnable;
-import org.apache.fluo.cluster.util.FluoYarnConfig;
-import org.apache.fluo.cluster.yarn.FluoTwillApp;
-import org.apache.fluo.cluster.yarn.TwillUtil;
import org.apache.fluo.core.client.FluoAdminImpl;
import org.apache.fluo.core.util.CuratorUtil;
import org.apache.hadoop.fs.Path;
@@ -185,7 +180,8 @@ public class YarnAppRunner extends ClusterAppRunner implements AutoCloseable {
throw new FluoException("Invalid fluo.properties due to " + e.getMessage(), e);
}
- TwillPreparer preparer = getTwillRunner(config).prepare(new FluoTwillApp(config, appConfDir));
+ TwillPreparer preparer = getTwillRunner(config)
+ .prepare(new org.apache.fluo.cluster.yarn.FluoTwillApp(config, appConfDir));
// Add jars from fluo lib/ directory that are not being loaded by Twill.
try {
@@ -327,17 +323,24 @@ public class YarnAppRunner extends ClusterAppRunner implements AutoCloseable {
}
private boolean allContainersRunning(TwillController controller, FluoConfiguration config) {
- return TwillUtil.numRunning(controller, OracleRunnable.ORACLE_NAME) == FluoYarnConfig
- .getOracleInstances(config)
- && TwillUtil.numRunning(controller, WorkerRunnable.WORKER_NAME) == FluoYarnConfig
- .getWorkerInstances(config);
+ return org.apache.fluo.cluster.yarn.TwillUtil.numRunning(controller,
+ org.apache.fluo.cluster.runnable.OracleRunnable.ORACLE_NAME) == org.apache.fluo.cluster.util.FluoYarnConfig
+ .getOracleInstances(config)
+ && org.apache.fluo.cluster.yarn.TwillUtil.numRunning(controller,
+ org.apache.fluo.cluster.runnable.WorkerRunnable.WORKER_NAME) == org.apache.fluo.cluster.util.FluoYarnConfig
+ .getWorkerInstances(config);
}
private String containerStatus(TwillController controller, FluoConfiguration config) {
- return "" + TwillUtil.numRunning(controller, OracleRunnable.ORACLE_NAME) + " of "
- + FluoYarnConfig.getOracleInstances(config) + " Oracle containers and "
- + TwillUtil.numRunning(controller, WorkerRunnable.WORKER_NAME) + " of "
- + FluoYarnConfig.getWorkerInstances(config) + " Worker containers";
+ return ""
+ + org.apache.fluo.cluster.yarn.TwillUtil.numRunning(controller,
+ org.apache.fluo.cluster.runnable.OracleRunnable.ORACLE_NAME)
+ + " of " + org.apache.fluo.cluster.util.FluoYarnConfig.getOracleInstances(config)
+ + " Oracle containers and "
+ + org.apache.fluo.cluster.yarn.TwillUtil.numRunning(controller,
+ org.apache.fluo.cluster.runnable.WorkerRunnable.WORKER_NAME)
+ + " of " + org.apache.fluo.cluster.util.FluoYarnConfig.getWorkerInstances(config)
+ + " Worker containers";
}
public void status(FluoConfiguration config, boolean extraInfo) {
@@ -366,15 +369,19 @@ public class YarnAppRunner extends ClusterAppRunner implements AutoCloseable {
if (extraInfo) {
ResourceReport report = getResourceReport(controller, 30000);
Collection<TwillRunResources> resources;
- resources = report.getRunnableResources(OracleRunnable.ORACLE_NAME);
+ resources = report
+ .getRunnableResources(org.apache.fluo.cluster.runnable.OracleRunnable.ORACLE_NAME);
System.out.println("\nThe application has " + resources.size() + " of "
- + FluoYarnConfig.getOracleInstances(config) + " desired Oracle containers:\n");
- TwillUtil.printResources(resources);
+ + org.apache.fluo.cluster.util.FluoYarnConfig.getOracleInstances(config)
+ + " desired Oracle containers:\n");
+ org.apache.fluo.cluster.yarn.TwillUtil.printResources(resources);
- resources = report.getRunnableResources(WorkerRunnable.WORKER_NAME);
+ resources = report
+ .getRunnableResources(org.apache.fluo.cluster.runnable.WorkerRunnable.WORKER_NAME);
System.out.println("\nThe application has " + resources.size() + " of "
- + FluoYarnConfig.getWorkerInstances(config) + " desired Worker containers:\n");
- TwillUtil.printResources(resources);
+ + org.apache.fluo.cluster.util.FluoYarnConfig.getWorkerInstances(config)
+ + " desired Worker containers:\n");
+ org.apache.fluo.cluster.yarn.TwillUtil.printResources(resources);
}
}
}
diff --git a/modules/cluster/src/main/java/org/apache/fluo/cluster/yarn/FluoTwillApp.java b/modules/cluster/src/main/java/org/apache/fluo/cluster/yarn/FluoTwillApp.java
index f0d8d73..eda580a 100644
--- a/modules/cluster/src/main/java/org/apache/fluo/cluster/yarn/FluoTwillApp.java
+++ b/modules/cluster/src/main/java/org/apache/fluo/cluster/yarn/FluoTwillApp.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -18,10 +18,6 @@ package org.apache.fluo.cluster.yarn;
import java.io.File;
import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.cluster.runnable.OracleRunnable;
-import org.apache.fluo.cluster.runnable.WorkerRunnable;
-import org.apache.fluo.cluster.runner.YarnAppRunner;
-import org.apache.fluo.cluster.util.FluoYarnConfig;
import org.apache.twill.api.ResourceSpecification;
import org.apache.twill.api.ResourceSpecification.SizeUnit;
import org.apache.twill.api.TwillApplication;
@@ -71,26 +67,37 @@ public class FluoTwillApp implements TwillApplication {
@Override
public TwillSpecification configure() {
- final int oracleInstances = FluoYarnConfig.getOracleInstances(config);
- final int oracleMaxMemory = FluoYarnConfig.getOracleMaxMemory(config);
- final int oracleNumCores = FluoYarnConfig.getOracleNumCores(config);
- final int workerInstances = FluoYarnConfig.getWorkerInstances(config);
- final int workerMaxMemory = FluoYarnConfig.getWorkerMaxMemory(config);
- final int workerNumCores = FluoYarnConfig.getWorkerNumCores(config);
+ final int oracleInstances =
+ org.apache.fluo.cluster.util.FluoYarnConfig.getOracleInstances(config);
+ final int oracleMaxMemory =
+ org.apache.fluo.cluster.util.FluoYarnConfig.getOracleMaxMemory(config);
+ final int oracleNumCores =
+ org.apache.fluo.cluster.util.FluoYarnConfig.getOracleNumCores(config);
+ final int workerInstances =
+ org.apache.fluo.cluster.util.FluoYarnConfig.getWorkerInstances(config);
+ final int workerMaxMemory =
+ org.apache.fluo.cluster.util.FluoYarnConfig.getWorkerMaxMemory(config);
+ final int workerNumCores =
+ org.apache.fluo.cluster.util.FluoYarnConfig.getWorkerNumCores(config);
log.info(
"Configuring Fluo '{}' application with {} Oracle instances and {} Worker instances "
+ "with following properties:",
config.getApplicationName(), oracleInstances, workerInstances);
- log.info("{} = {}", FluoYarnConfig.ORACLE_MAX_MEMORY_MB_PROP, oracleMaxMemory);
- log.info("{} = {}", FluoYarnConfig.WORKER_MAX_MEMORY_MB_PROP, workerMaxMemory);
- log.info("{} = {}", FluoYarnConfig.ORACLE_NUM_CORES_PROP, oracleNumCores);
- log.info("{} = {}", FluoYarnConfig.WORKER_NUM_CORES_PROP, workerNumCores);
+ log.info("{} = {}", org.apache.fluo.cluster.util.FluoYarnConfig.ORACLE_MAX_MEMORY_MB_PROP,
+ oracleMaxMemory);
+ log.info("{} = {}", org.apache.fluo.cluster.util.FluoYarnConfig.WORKER_MAX_MEMORY_MB_PROP,
+ workerMaxMemory);
+ log.info("{} = {}", org.apache.fluo.cluster.util.FluoYarnConfig.ORACLE_NUM_CORES_PROP,
+ oracleNumCores);
+ log.info("{} = {}", org.apache.fluo.cluster.util.FluoYarnConfig.WORKER_NUM_CORES_PROP,
+ workerNumCores);
// Start building Fluo Twill application
- MoreRunnable moreRunnable = TwillSpecification.Builder.with()
- .setName(YarnAppRunner.getYarnApplicationName(config.getApplicationName())).withRunnable();
+ MoreRunnable moreRunnable =
+ TwillSpecification.Builder.with().setName(org.apache.fluo.cluster.runner.YarnAppRunner
+ .getYarnApplicationName(config.getApplicationName())).withRunnable();
// Configure Oracle(s)
ResourceSpecification oracleResources =
@@ -98,7 +105,9 @@ public class FluoTwillApp implements TwillApplication {
.setMemory(oracleMaxMemory, SizeUnit.MEGA).setInstances(oracleInstances).build();
LocalFileAdder fileAdder = moreRunnable
- .add(OracleRunnable.ORACLE_NAME, new OracleRunnable(), oracleResources).withLocalFiles();
+ .add(org.apache.fluo.cluster.runnable.OracleRunnable.ORACLE_NAME,
+ new org.apache.fluo.cluster.runnable.OracleRunnable(), oracleResources)
+ .withLocalFiles();
RunnableSetter runnableSetter = addConfigFiles(fileAdder).apply();
// Configure Worker(s)
@@ -107,11 +116,14 @@ public class FluoTwillApp implements TwillApplication {
.setMemory(workerMaxMemory, SizeUnit.MEGA).setInstances(workerInstances).build();
fileAdder = runnableSetter
- .add(WorkerRunnable.WORKER_NAME, new WorkerRunnable(), workerResources).withLocalFiles();
+ .add(org.apache.fluo.cluster.runnable.WorkerRunnable.WORKER_NAME,
+ new org.apache.fluo.cluster.runnable.WorkerRunnable(), workerResources)
+ .withLocalFiles();
runnableSetter = addConfigFiles(fileAdder).apply();
// Set runnable order, build and return TwillSpecification
- return runnableSetter.withOrder().begin(OracleRunnable.ORACLE_NAME)
- .nextWhenStarted(WorkerRunnable.WORKER_NAME).build();
+ return runnableSetter.withOrder()
+ .begin(org.apache.fluo.cluster.runnable.OracleRunnable.ORACLE_NAME)
+ .nextWhenStarted(org.apache.fluo.cluster.runnable.WorkerRunnable.WORKER_NAME).build();
}
}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
index f25a110..fc2701f 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -23,7 +23,6 @@ import java.net.URI;
import java.net.URL;
import java.net.URLClassLoader;
import java.nio.charset.StandardCharsets;
-import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
@@ -52,7 +51,6 @@ import org.apache.fluo.api.exceptions.FluoException;
import org.apache.fluo.core.impl.FluoConfigurationImpl;
import org.apache.fluo.core.observer.ObserverUtil;
import org.apache.fluo.core.util.AccumuloUtil;
-import org.apache.fluo.core.util.ByteUtil;
import org.apache.fluo.core.util.CuratorUtil;
import org.apache.fluo.core.worker.finder.hash.PartitionManager;
import org.apache.hadoop.conf.Configuration;
@@ -160,7 +158,9 @@ public class FluoAdminImpl implements FluoAdmin {
if (!accumuloJars.isEmpty()) {
accumuloClasspath = copyJarsToDfs(accumuloJars, "lib/accumulo");
} else {
- accumuloClasspath = config.getAccumuloClasspath().trim();
+ @SuppressWarnings("deprecation")
+ String tmpCP = config.getAccumuloClasspath().trim();
+ accumuloClasspath = tmpCP;
}
if (!accumuloClasspath.isEmpty()) {
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java b/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java
index 4172386..ea00c73 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/LockResolver.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -87,7 +87,6 @@ public class LockResolver {
return groupedLocks;
}
-
private static class LockInfo {
final Bytes prow;
@@ -125,7 +124,7 @@ public class LockResolver {
* @param startTs The logical start time from the oracle of the transaction that encountered the
* lock
* @param stats Stats object for the transaction that encountered the lock
- * @param locks List of locks
+ * @param locksKVs List of locks
* @param startTime The wall time that the transaction that encountered the lock first saw the
* lock
* @return true if all locks passed in were resolved (rolled forward or back)
diff --git a/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserverStoreV1.java b/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserverStoreV1.java
index feba76e..6c0f27f 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserverStoreV1.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserverStoreV1.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -30,13 +30,11 @@ import com.google.common.collect.ImmutableMap;
import org.apache.curator.framework.CuratorFramework;
import org.apache.fluo.accumulo.util.ZookeeperPath;
import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Column;
import org.apache.fluo.api.exceptions.FluoException;
import org.apache.fluo.api.observer.Observer;
import org.apache.fluo.api.observer.Observer.NotificationType;
-import org.apache.fluo.api.observer.Observer.ObservedColumn;
import org.apache.fluo.core.impl.Environment;
import org.apache.fluo.core.observer.ObserverStore;
import org.apache.fluo.core.observer.Observers;
@@ -58,18 +56,20 @@ public class ObserverStoreV1 implements ObserverStore {
@Override
public boolean handles(FluoConfiguration config) {
- Collection<ObserverSpecification> obsSpecs = config.getObserverSpecifications();
+ Collection<org.apache.fluo.api.config.ObserverSpecification> obsSpecs =
+ config.getObserverSpecifications();
return !obsSpecs.isEmpty();
}
@Override
public void update(CuratorFramework curator, FluoConfiguration config) throws Exception {
- Collection<ObserverSpecification> obsSpecs = config.getObserverSpecifications();
+ Collection<org.apache.fluo.api.config.ObserverSpecification> obsSpecs =
+ config.getObserverSpecifications();
- Map<Column, ObserverSpecification> colObservers = new HashMap<>();
- Map<Column, ObserverSpecification> weakObservers = new HashMap<>();
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> colObservers = new HashMap<>();
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> weakObservers = new HashMap<>();
- for (ObserverSpecification ospec : obsSpecs) {
+ for (org.apache.fluo.api.config.ObserverSpecification ospec : obsSpecs) {
Observer observer;
try {
observer = Class.forName(ospec.getClassName()).asSubclass(Observer.class).newInstance();
@@ -92,7 +92,8 @@ public class ObserverStoreV1 implements ObserverStore {
e);
}
- ObservedColumn observedCol = observer.getObservedColumn();
+ org.apache.fluo.api.observer.Observer.ObservedColumn observedCol =
+ observer.getObservedColumn();
if (observedCol.getType() == NotificationType.STRONG) {
colObservers.put(observedCol.getColumn(), ospec);
} else {
@@ -104,8 +105,9 @@ public class ObserverStoreV1 implements ObserverStore {
}
private static void updateObservers(CuratorFramework curator,
- Map<Column, ObserverSpecification> colObservers,
- Map<Column, ObserverSpecification> weakObservers) throws Exception {
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> colObservers,
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> weakObservers)
+ throws Exception {
// TODO check that no workers are running... or make workers watch this znode
@@ -126,14 +128,16 @@ public class ObserverStoreV1 implements ObserverStore {
}
private static void serializeObservers(DataOutputStream dos,
- Map<Column, ObserverSpecification> colObservers) throws IOException {
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> colObservers)
+ throws IOException {
// TODO use a human readable serialized format like json
- Set<Entry<Column, ObserverSpecification>> es = colObservers.entrySet();
+ Set<Entry<Column, org.apache.fluo.api.config.ObserverSpecification>> es =
+ colObservers.entrySet();
WritableUtils.writeVInt(dos, colObservers.size());
- for (Entry<Column, ObserverSpecification> entry : es) {
+ for (Entry<Column, org.apache.fluo.api.config.ObserverSpecification> entry : es) {
ColumnUtil.writeColumn(entry.getKey(), dos);
dos.writeUTF(entry.getValue().getClassName());
Map<String, String> params = entry.getValue().getConfiguration().toMap();
@@ -145,8 +149,10 @@ public class ObserverStoreV1 implements ObserverStore {
}
}
- private static byte[] serializeObservers(Map<Column, ObserverSpecification> colObservers,
- Map<Column, ObserverSpecification> weakObservers) throws IOException {
+ private static byte[] serializeObservers(
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> colObservers,
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> weakObservers)
+ throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (DataOutputStream dos = new DataOutputStream(baos)) {
serializeObservers(dos, colObservers);
@@ -157,12 +163,11 @@ public class ObserverStoreV1 implements ObserverStore {
return serializedObservers;
}
+ private static Map<Column, org.apache.fluo.api.config.ObserverSpecification> readObservers(
+ DataInputStream dis) throws IOException {
- private static Map<Column, ObserverSpecification> readObservers(DataInputStream dis)
- throws IOException {
-
- ImmutableMap.Builder<Column, ObserverSpecification> omapBuilder =
- new ImmutableMap.Builder<Column, ObserverSpecification>();
+ ImmutableMap.Builder<Column, org.apache.fluo.api.config.ObserverSpecification> omapBuilder =
+ new ImmutableMap.Builder<>();
int num = WritableUtils.readVInt(dis);
for (int i = 0; i < num; i++) {
@@ -176,7 +181,8 @@ public class ObserverStoreV1 implements ObserverStore {
params.put(k, v);
}
- ObserverSpecification ospec = new ObserverSpecification(clazz, params);
+ org.apache.fluo.api.config.ObserverSpecification ospec =
+ new org.apache.fluo.api.config.ObserverSpecification(clazz, params);
omapBuilder.put(col, ospec);
}
return omapBuilder.build();
@@ -185,8 +191,8 @@ public class ObserverStoreV1 implements ObserverStore {
@Override
public RegisteredObservers load(CuratorFramework curator) throws Exception {
- Map<Column, ObserverSpecification> observers;
- Map<Column, ObserverSpecification> weakObservers;
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> observers;
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> weakObservers;
ByteArrayInputStream bais;
try {
@@ -200,7 +206,6 @@ public class ObserverStoreV1 implements ObserverStore {
observers = readObservers(dis);
weakObservers = readObservers(dis);
-
return new RegisteredObservers() {
@Override
diff --git a/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserversV1.java b/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserversV1.java
index ff3c667..3965296 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserversV1.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/observer/v1/ObserversV1.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -22,7 +22,6 @@ import java.util.Map;
import java.util.Map.Entry;
import com.google.common.collect.Iterables;
-import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.api.data.Column;
import org.apache.fluo.api.observer.Observer;
import org.apache.fluo.core.impl.Environment;
@@ -37,8 +36,8 @@ class ObserversV1 implements Observers {
private Environment env;
Map<Column, List<Observer>> observers = new HashMap<>();
- Map<Column, ObserverSpecification> strongObservers;
- Map<Column, ObserverSpecification> weakObservers;
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> strongObservers;
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> weakObservers;
Map<Column, String> aliases;
private List<Observer> getObserverList(Column col) {
@@ -53,16 +52,17 @@ class ObserversV1 implements Observers {
return observerList;
}
- public ObserversV1(Environment env, Map<Column, ObserverSpecification> strongObservers,
- Map<Column, ObserverSpecification> weakObservers) {
+ public ObserversV1(Environment env,
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> strongObservers,
+ Map<Column, org.apache.fluo.api.config.ObserverSpecification> weakObservers) {
this.env = env;
this.strongObservers = strongObservers;
this.weakObservers = weakObservers;
this.aliases = new HashMap<>();
- for (Entry<Column, ObserverSpecification> e : Iterables.concat(strongObservers.entrySet(),
- weakObservers.entrySet())) {
- ObserverSpecification observerConfig = e.getValue();
+ for (Entry<Column, org.apache.fluo.api.config.ObserverSpecification> e : Iterables
+ .concat(strongObservers.entrySet(), weakObservers.entrySet())) {
+ org.apache.fluo.api.config.ObserverSpecification observerConfig = e.getValue();
try {
String alias =
Class.forName(observerConfig.getClassName()).asSubclass(Observer.class).getSimpleName();
@@ -73,6 +73,7 @@ class ObserversV1 implements Observers {
}
}
+ @Override
public Observer getObserver(Column col) {
List<Observer> observerList;
@@ -86,7 +87,7 @@ class ObserversV1 implements Observers {
Observer observer = null;
- ObserverSpecification observerConfig = strongObservers.get(col);
+ org.apache.fluo.api.config.ObserverSpecification observerConfig = strongObservers.get(col);
if (observerConfig == null) {
observerConfig = weakObservers.get(col);
}
@@ -112,6 +113,7 @@ class ObserversV1 implements Observers {
return observer;
}
+ @Override
public void returnObserver(Observer observer) {
List<Observer> olist = getObserverList(observer.getObservedColumn().getColumn());
synchronized (olist) {
diff --git a/modules/core/src/main/java/org/apache/fluo/core/observer/v2/ObserverStoreV2.java b/modules/core/src/main/java/org/apache/fluo/core/observer/v2/ObserverStoreV2.java
index c7ac734..c5731d8 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/observer/v2/ObserverStoreV2.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/observer/v2/ObserverStoreV2.java
@@ -105,8 +105,8 @@ public class ObserverStoreV2 implements ObserverStore {
String json = new String(data, UTF_8);
JsonObservers jco = new Gson().fromJson(json, JsonObservers.class);
- ImmutableSet.Builder<Column> weakColumnsBuilder = new ImmutableSet.Builder<Column>();
- ImmutableSet.Builder<Column> strongColumnsBuilder = new ImmutableSet.Builder<Column>();
+ ImmutableSet.Builder<Column> weakColumnsBuilder = new ImmutableSet.Builder<>();
+ ImmutableSet.Builder<Column> strongColumnsBuilder = new ImmutableSet.Builder<>();
for (Entry<Column, NotificationType> entry : jco.getObservedColumns().entrySet()) {
switch (entry.getValue()) {
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java
index e33b28b..22842bc 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -90,6 +90,11 @@ class PartitionInfo {
}
@Override
+ public int hashCode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
public String toString() {
return String.format(
"workers:%d groups:%d groupSize:%d groupId:%d idInGroup:%d #tablets:%d", numWorkers,
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java
index 40dc5e4..2957f7b 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -75,6 +75,11 @@ public class RangeSet {
return false;
}
+ @Override
+ public int hashCode() {
+ throw new UnsupportedOperationException();
+ }
+
public int size() {
return tmap.size() + (lastRange == null ? 0 : 1);
}
diff --git a/modules/core/src/test/java/org/apache/fluo/core/observer/ObserverStoreTest.java b/modules/core/src/test/java/org/apache/fluo/core/observer/ObserverStoreTest.java
index fd94a12..c6fe242 100644
--- a/modules/core/src/test/java/org/apache/fluo/core/observer/ObserverStoreTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/core/observer/ObserverStoreTest.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -16,7 +16,6 @@
package org.apache.fluo.core.observer;
import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.core.observer.v1.ObserverStoreV1;
import org.apache.fluo.core.observer.v2.ObserverStoreV2;
import org.junit.Assert;
@@ -40,7 +39,7 @@ public class ObserverStoreTest {
Assert.assertTrue(ov2.handles(config));
config = new FluoConfiguration();
- config.addObserver(new ObserverSpecification("TestProvider2"));
+ config.addObserver(new org.apache.fluo.api.config.ObserverSpecification("TestProvider2"));
Assert.assertTrue(ov1.handles(config));
Assert.assertFalse(ov2.handles(config));
}
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ObserverConfigIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ObserverConfigIT.java
index c29de58..e792868 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ObserverConfigIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ObserverConfigIT.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -24,13 +24,11 @@ import org.apache.fluo.api.client.Snapshot;
import org.apache.fluo.api.client.Transaction;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
import org.apache.fluo.api.data.Column;
import org.apache.fluo.api.metrics.Counter;
import org.apache.fluo.api.metrics.Meter;
-import org.apache.fluo.api.observer.AbstractObserver;
import org.apache.fluo.api.observer.Observer.NotificationType;
import org.apache.fluo.integration.ITBaseMini;
import org.junit.Assert;
@@ -41,7 +39,7 @@ import org.junit.rules.Timeout;
@Deprecated
public class ObserverConfigIT extends ITBaseMini {
- public static class ConfigurableObserver extends AbstractObserver {
+ public static class ConfigurableObserver extends org.apache.fluo.api.observer.AbstractObserver {
@Rule
public Timeout globalTimeout = Timeout.seconds(getTestTimeout());
@@ -101,17 +99,20 @@ public class ObserverConfigIT extends ITBaseMini {
@Override
protected void setupObservers(FluoConfiguration fc) {
- List<ObserverSpecification> observers = new ArrayList<>();
+ List<org.apache.fluo.api.config.ObserverSpecification> observers = new ArrayList<>();
- observers.add(new ObserverSpecification(ConfigurableObserver.class.getName(),
- newMap("observedCol", "fam1:col1:" + NotificationType.STRONG, "outputCQ", "col2")));
+ observers.add(
+ new org.apache.fluo.api.config.ObserverSpecification(ConfigurableObserver.class.getName(),
+ newMap("observedCol", "fam1:col1:" + NotificationType.STRONG, "outputCQ", "col2")));
- observers.add(new ObserverSpecification(ConfigurableObserver.class.getName(),
- newMap("observedCol", "fam1:col2:" + NotificationType.STRONG, "outputCQ", "col3",
- "setWeakNotification", "true")));
+ observers.add(
+ new org.apache.fluo.api.config.ObserverSpecification(ConfigurableObserver.class.getName(),
+ newMap("observedCol", "fam1:col2:" + NotificationType.STRONG, "outputCQ", "col3",
+ "setWeakNotification", "true")));
- observers.add(new ObserverSpecification(ConfigurableObserver.class.getName(),
- newMap("observedCol", "fam1:col3:" + NotificationType.WEAK, "outputCQ", "col4")));
+ observers.add(
+ new org.apache.fluo.api.config.ObserverSpecification(ConfigurableObserver.class.getName(),
+ newMap("observedCol", "fam1:col3:" + NotificationType.WEAK, "outputCQ", "col4")));
fc.addObservers(observers);
}
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java
index 489b9e0..24d7cd4 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ReadLockFailureIT.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -18,7 +18,6 @@ package org.apache.fluo.integration.impl;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.function.Consumer;
@@ -26,8 +25,6 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.fluo.accumulo.format.FluoFormatter;
import org.apache.fluo.api.client.Snapshot;
@@ -51,13 +48,6 @@ import static org.apache.fluo.integration.impl.ReadLockIT.setAlias;
public class ReadLockFailureIT extends ITBaseImpl {
- private void dumpTable(Consumer<String> out) throws TableNotFoundException {
- Scanner scanner = conn.createScanner(getCurTableName(), Authorizations.EMPTY);
- for (Entry<Key, Value> entry : scanner) {
- out.accept(FluoFormatter.toString(entry));
- }
- }
-
private Set<String> getDerivedEdges() {
Set<String> derivedEdges = new HashSet<>();
try (Snapshot snap = client.newSnapshot()) {
@@ -98,7 +88,6 @@ public class ReadLockFailureIT extends ITBaseImpl {
}
}
-
private TransactorNode partiallyCommit(Consumer<TransactionBase> action, boolean commitPrimary,
boolean closeTransactor) throws Exception {
TransactorNode t2 = new TransactorNode(env);
@@ -223,7 +212,7 @@ public class ReadLockFailureIT extends ITBaseImpl {
}, false, closeTransactor);
retryTwice(tx -> {
- Map<String, Map<Column, String>> ratios = tx.gets(Arrays.asList("user5", "user6"), crCol);
+ tx.gets(Arrays.asList("user5", "user6"), crCol);
tx.set("user5", crCol, "0.51");
tx.set("user6", crCol, "0.76");
@@ -267,7 +256,6 @@ public class ReadLockFailureIT extends ITBaseImpl {
Map<RowColumn, String> ratios = tx.withReadLock()
.gets(Arrays.asList(new RowColumn("user5", crCol), new RowColumn("user6", crCol)));
-
double cr1 = Double.parseDouble(ratios.get(new RowColumn("user5", crCol)));
double cr2 = Double.parseDouble(ratios.get(new RowColumn("user6", crCol)));
@@ -275,8 +263,7 @@ public class ReadLockFailureIT extends ITBaseImpl {
}, false, true);
retryTwice(tx -> {
- Map<RowColumn, String> ratios =
- tx.gets(Arrays.asList(new RowColumn("user5", crCol), new RowColumn("user6", crCol)));
+ tx.gets(Arrays.asList(new RowColumn("user5", crCol), new RowColumn("user6", crCol)));
tx.set("user5", crCol, "0.51");
tx.set("user6", crCol, "0.76");
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
index bd6d5cc..d08730f 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
@@ -39,8 +39,7 @@ import org.junit.rules.Timeout;
import static org.apache.fluo.api.observer.Observer.NotificationType.STRONG;
/**
- * A simple test that added links between nodes in a graph. There is an observer that updates an
- * index of node degree.
+ * A simple test that added links between nodes in a graph. There is an observer that updates an index of node degree.
*/
public class WorkerIT extends ITBaseMini {
@Rule
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ZKSecretIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ZKSecretIT.java
index 1605d2b..9786fd1 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ZKSecretIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ZKSecretIT.java
@@ -4,9 +4,9 @@
* copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
@@ -85,7 +85,7 @@ public class ZKSecretIT extends ITBaseMini {
FluoConfiguration conf = new FluoConfiguration(miniFluo.getClientConfiguration());
conf.setZookeeperSecret("");
try (FluoClient client = FluoFactory.newClient(conf)) {
- Assert.fail("Expected client creation to fail.");
+ Assert.fail("Expected client creation to fail. " + client);
} catch (Exception e) {
boolean sawNoAuth = false;
Throwable throwable = e;
@@ -121,7 +121,6 @@ public class ZKSecretIT extends ITBaseMini {
ZooKeeper zk = getZookeeper();
-
// Verify oracle gc timestamp is visible w/o a password. The GC iterator that runs in Accumulo
// tablet servers reads this.
String ts = new String(zk.getData(ZookeeperPath.ORACLE_GC_TIMESTAMP, false, null),
@@ -151,7 +150,6 @@ public class ZKSecretIT extends ITBaseMini {
} catch (NoAuthException nae) {
}
-
try {
zk.getChildren(path, false);
} catch (NoAuthException nae) {
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java
index 0835310..900592c 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/log/LogIT.java
@@ -328,7 +328,6 @@ public class LogIT extends ITBaseMini {
tx.commit();
}
-
Logger logger = Logger.getLogger("fluo.tx");
StringWriter writer = new StringWriter();
--
To stop receiving notification emails like this one, please contact
"commits@fluo.apache.org" <co...@fluo.apache.org>.