This is an automated email from the ASF dual-hosted git repository. mmiklavcic pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/metron.git
The following commit(s) were added to refs/heads/master by this push: new c402e64 METRON-2217 Migrate current HBase client from HTableInterface to Table (mmiklavc) closes apache/metron#1483 c402e64 is described below commit c402e6442d8d699eef5aa5464cd91347160b082c Author: mmiklavc <michael.miklav...@gmail.com> AuthorDate: Wed Sep 4 14:29:46 2019 -0600 METRON-2217 Migrate current HBase client from HTableInterface to Table (mmiklavc) closes apache/metron#1483 --- .../profiler/client/HBaseProfilerClient.java | 34 +++++++----- .../metron/profiler/client/stellar/GetProfile.java | 61 ++++++++-------------- .../profiler/client/stellar/VerboseProfile.java | 57 ++++++++------------ .../profiler/client/HBaseProfilerClientTest.java | 30 ++++++----- .../metron/profiler/client/ProfileWriter.java | 25 +++++---- .../profiler/client/stellar/GetProfileTest.java | 35 ++++++------- .../client/stellar/VerboseProfileTest.java | 50 +++++++++--------- .../metron/profiler/spark/BatchProfilerConfig.java | 11 ++-- .../org/apache/metron/rest/config/IndexConfig.java | 2 +- .../metron/rest/user/UserSettingsClient.java | 8 +-- .../apache/metron/rest/config/HBaseConfigTest.java | 8 --- .../org/apache/metron/rest/config/TestConfig.java | 4 +- .../metron/rest/user/UserSettingsClientTest.java | 35 +++++++------ .../metron/dataloads/hbase/mr/PrunerMapper.java | 3 +- .../nonbulk/flatfile/HBaseExtractorState.java | 8 +-- .../nonbulk/flatfile/importer/LocalImporter.java | 12 +++-- .../dataloads/nonbulk/taxii/TaxiiHandler.java | 16 +++--- .../mr/LeastRecentlyUsedPrunerIntegrationTest.java | 22 ++++---- ...pleEnrichmentFlatFileLoaderIntegrationTest.java | 35 ++++++------- .../nonbulk/taxii/TaxiiIntegrationTest.java | 4 +- .../enrichment/adapters/cif/CIFHbaseAdapter.java | 13 ++--- .../adapters/simplehbase/SimpleHBaseAdapter.java | 2 + .../adapters/threatintel/ThreatIntelAdapter.java | 2 + .../enrichment/converter/EnrichmentHelper.java | 7 ++- .../metron/enrichment/lookup/EnrichmentLookup.java | 18 +++---- .../lookup/accesstracker/AccessTrackerUtil.java | 4 +- .../accesstracker/PersistentAccessTracker.java | 8 +-- .../PersistentBloomTrackerCreator.java | 9 ++-- .../stellar/SimpleHBaseEnrichmentFunctions.java | 20 +++---- .../metron/enrichment/utils/EnrichmentUtils.java | 6 +-- .../threatintel/ThreatIntelAdapterTest.java | 4 +- .../EnrichmentCoprocessorIntegrationTest.java | 11 ++-- .../coprocessor/EnrichmentCoprocessorTest.java | 5 +- .../org/apache/metron/hbase/HTableProvider.java | 46 ++++++++++++++-- .../org/apache/metron/hbase/TableProvider.java | 8 +-- .../apache/metron/hbase/client/HBaseClient.java | 11 ++-- .../metron/hbase/client/HBaseClientTest.java | 37 +++++++------ .../metron/hbase/mock/MockHBaseTableProvider.java | 18 +++---- .../org/apache/metron/hbase/mock/MockHTable.java | 35 ++++--------- .../org/apache/metron/indexing/dao/HBaseDao.java | 6 +-- .../writer/hbase/SimpleHbaseEnrichmentWriter.java | 12 ++--- 41 files changed, 372 insertions(+), 370 deletions(-) diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/HBaseProfilerClient.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/HBaseProfilerClient.java index 2e537da..f4bff2b 100644 --- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/HBaseProfilerClient.java +++ b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/HBaseProfilerClient.java @@ -20,22 +20,22 @@ package org.apache.metron.profiler.client; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.util.Bytes; import org.apache.metron.common.utils.SerDeUtils; +import org.apache.metron.hbase.TableProvider; import org.apache.metron.profiler.ProfileMeasurement; import org.apache.metron.profiler.ProfilePeriod; import org.apache.metron.profiler.hbase.ColumnBuilder; import org.apache.metron.profiler.hbase.RowKeyBuilder; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.Optional; -import java.util.concurrent.TimeUnit; - /** * The default implementation of a ProfilerClient that fetches profile data persisted in HBase. */ @@ -44,7 +44,7 @@ public class HBaseProfilerClient implements ProfilerClient { /** * Used to access the profile data stored in HBase. */ - private HTableInterface table; + private TableProvider tableProvider; /** * Generates the row keys necessary to scan HBase. @@ -57,15 +57,21 @@ public class HBaseProfilerClient implements ProfilerClient { private ColumnBuilder columnBuilder; private long periodDurationMillis; + private String tableName; + private final Configuration hbaseConfig; - public HBaseProfilerClient(HTableInterface table, + public HBaseProfilerClient(TableProvider tableProvider, RowKeyBuilder rowKeyBuilder, ColumnBuilder columnBuilder, - long periodDurationMillis) { - setTable(table); + long periodDurationMillis, + String tableName, + Configuration hbaseConfig) { + setTableProvider(tableProvider); setRowKeyBuilder(rowKeyBuilder); setColumnBuilder(columnBuilder); this.periodDurationMillis = periodDurationMillis; + this.tableName = tableName; + this.hbaseConfig = hbaseConfig; } /** @@ -135,7 +141,7 @@ public class HBaseProfilerClient implements ProfilerClient { // query HBase try { - Result[] results = table.get(gets); + Result[] results = tableProvider.getTable(hbaseConfig, tableName).get(gets); for(int i = 0; i < results.length; ++i) { Result result = results[i]; ProfileMeasurement measurement = measurements.get(i); @@ -164,8 +170,8 @@ public class HBaseProfilerClient implements ProfilerClient { } - public void setTable(HTableInterface table) { - this.table = table; + public void setTableProvider(TableProvider tableProvider) { + this.tableProvider = tableProvider; } public void setRowKeyBuilder(RowKeyBuilder rowKeyBuilder) { diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/GetProfile.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/GetProfile.java index a0e2bdf..405dd8b 100644 --- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/GetProfile.java +++ b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/GetProfile.java @@ -20,8 +20,25 @@ package org.apache.metron.profiler.client.stellar; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; +import static org.apache.metron.profiler.client.stellar.Util.getArg; +import static org.apache.metron.profiler.client.stellar.Util.getEffectiveConfig; +import static org.apache.metron.profiler.client.stellar.Util.getPeriodDurationInMillis; + +import java.lang.invoke.MethodHandles; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.metron.hbase.HTableProvider; import org.apache.metron.hbase.TableProvider; import org.apache.metron.profiler.ProfileMeasurement; @@ -39,25 +56,6 @@ import org.apache.metron.stellar.dsl.StellarFunction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.lang.invoke.MethodHandles; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.TimeUnit; - -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; -import static org.apache.metron.profiler.client.stellar.Util.getArg; -import static org.apache.metron.profiler.client.stellar.Util.getEffectiveConfig; -import static org.apache.metron.profiler.client.stellar.Util.getPeriodDurationInMillis; - /** * A Stellar function that can retrieve data contained within a Profile. * @@ -170,9 +168,10 @@ public class GetProfile implements StellarFunction { if (client == null || !cachedConfigMap.equals(effectiveConfig)) { RowKeyBuilder rowKeyBuilder = getRowKeyBuilder(effectiveConfig); ColumnBuilder columnBuilder = getColumnBuilder(effectiveConfig); - HTableInterface table = getTable(effectiveConfig); long periodDuration = getPeriodDurationInMillis(effectiveConfig); - client = new HBaseProfilerClient(table, rowKeyBuilder, columnBuilder, periodDuration); + String tableName = PROFILER_HBASE_TABLE.get(effectiveConfig, String.class); + Configuration hbaseConfig = HBaseConfiguration.create(); + client = new HBaseProfilerClient(getTableProvider(effectiveConfig), rowKeyBuilder, columnBuilder, periodDuration, tableName, hbaseConfig); cachedConfigMap = effectiveConfig; } if(cachedConfigMap != null) { @@ -248,24 +247,6 @@ public class GetProfile implements StellarFunction { } /** - * Create an HBase table used when accessing HBase. - * @param global The global configuration. - * @return - */ - private HTableInterface getTable(Map<String, Object> global) { - - String tableName = PROFILER_HBASE_TABLE.get(global, String.class); - TableProvider provider = getTableProvider(global); - - try { - return provider.getTable(HBaseConfiguration.create(), tableName); - - } catch (IOException e) { - throw new IllegalArgumentException(String.format("Unable to access table: %s", tableName), e); - } - } - - /** * Create the TableProvider to use when accessing HBase. * @param global The global configuration. */ diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/VerboseProfile.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/VerboseProfile.java index 9e857aa..0979525 100644 --- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/VerboseProfile.java +++ b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/VerboseProfile.java @@ -18,8 +18,24 @@ package org.apache.metron.profiler.client.stellar; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_DEFAULT_VALUE; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; +import static org.apache.metron.profiler.client.stellar.Util.getArg; +import static org.apache.metron.profiler.client.stellar.Util.getPeriodDurationInMillis; +import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG; + +import java.lang.invoke.MethodHandles; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.metron.hbase.HTableProvider; import org.apache.metron.hbase.TableProvider; import org.apache.metron.profiler.ProfileMeasurement; @@ -37,25 +53,6 @@ import org.apache.metron.stellar.dsl.StellarFunction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.lang.invoke.MethodHandles; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.TimeUnit; - -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_DEFAULT_VALUE; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; -import static org.apache.metron.profiler.client.stellar.Util.getArg; -import static org.apache.metron.profiler.client.stellar.Util.getPeriodDurationInMillis; -import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG; - /** * A Stellar function that can retrieve profile measurements. * @@ -130,9 +127,9 @@ public class VerboseProfile implements StellarFunction { if (client == null) { RowKeyBuilder rowKeyBuilder = getRowKeyBuilder(globals); ColumnBuilder columnBuilder = getColumnBuilder(globals); - HTableInterface table = getTable(globals); + TableProvider provider = getTableProvider(globals); long periodDuration = getPeriodDurationInMillis(globals); - client = new HBaseProfilerClient(table, rowKeyBuilder, columnBuilder, periodDuration); + client = new HBaseProfilerClient(provider, rowKeyBuilder, columnBuilder, periodDuration, getTableName(globals), HBaseConfiguration.create()); } // is there a default value? @@ -185,20 +182,8 @@ public class VerboseProfile implements StellarFunction { return new SaltyRowKeyBuilder(saltDivisor, getPeriodDurationInMillis(global), TimeUnit.MILLISECONDS); } - /** - * Create an HBase table used when accessing HBase. - * @param global The global configuration. - * @return - */ - private HTableInterface getTable(Map<String, Object> global) { - String tableName = PROFILER_HBASE_TABLE.get(global, String.class); - TableProvider provider = getTableProvider(global); - try { - return provider.getTable(HBaseConfiguration.create(), tableName); - - } catch (IOException e) { - throw new IllegalArgumentException(String.format("Unable to access table: %s", tableName), e); - } + private String getTableName(Map<String, Object> global) { + return PROFILER_HBASE_TABLE.get(global, String.class); } /** diff --git a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/HBaseProfilerClientTest.java b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/HBaseProfilerClientTest.java index cc3748e..c7ad11a 100644 --- a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/HBaseProfilerClientTest.java +++ b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/HBaseProfilerClientTest.java @@ -20,6 +20,13 @@ package org.apache.metron.profiler.client; +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import org.apache.metron.hbase.mock.MockHBaseTableProvider; import org.apache.metron.hbase.mock.MockHTable; import org.apache.metron.profiler.ProfileMeasurement; import org.apache.metron.profiler.hbase.ColumnBuilder; @@ -32,14 +39,6 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; -import java.util.Arrays; -import java.util.List; -import java.util.Optional; -import java.util.concurrent.TimeUnit; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - /** * Tests the HBaseProfilerClient. * @@ -59,26 +58,29 @@ public class HBaseProfilerClientTest { private HBaseProfilerClient client; private StellarStatefulExecutor executor; - private MockHTable table; + private MockHBaseTableProvider provider; private ProfileWriter profileWriter; @Before public void setup() throws Exception { - table = new MockHTable(tableName, columnFamily); + provider = new MockHBaseTableProvider(); executor = new DefaultStellarStatefulExecutor(); + MockHBaseTableProvider.addToCache(tableName, columnFamily); // writes values to be read during testing long periodDurationMillis = periodUnits.toMillis(periodDuration); RowKeyBuilder rowKeyBuilder = new SaltyRowKeyBuilder(); ColumnBuilder columnBuilder = new ValueOnlyColumnBuilder(columnFamily); - profileWriter = new ProfileWriter(rowKeyBuilder, columnBuilder, table, periodDurationMillis); + profileWriter = new ProfileWriter(rowKeyBuilder, columnBuilder, provider, periodDurationMillis, + tableName, null); - client = new HBaseProfilerClient(table, rowKeyBuilder, columnBuilder, periodDurationMillis); + client = new HBaseProfilerClient(provider, rowKeyBuilder, columnBuilder, periodDurationMillis, + tableName, null); } @After public void tearDown() throws Exception { - table.clear(); + ((MockHTable) provider.getTable(null, tableName)).clear(); } @Test @@ -200,4 +202,4 @@ public class HBaseProfilerClientTest { List<ProfileMeasurement> results = client.fetch(Integer.class, profile, entity, group, startFetchAt, endFetchAt, Optional.empty()); assertEquals(0, results.size()); } -} \ No newline at end of file +} diff --git a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/ProfileWriter.java b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/ProfileWriter.java index 4e00164..c2fa4e7 100644 --- a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/ProfileWriter.java +++ b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/ProfileWriter.java @@ -20,12 +20,17 @@ package org.apache.metron.profiler.client; +import java.util.Collections; +import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Durability; -import org.apache.hadoop.hbase.client.HTableInterface; -import org.apache.metron.hbase.HTableProvider; import org.apache.metron.hbase.ColumnList; +import org.apache.metron.hbase.HTableProvider; +import org.apache.metron.hbase.TableProvider; import org.apache.metron.hbase.client.HBaseClient; import org.apache.metron.profiler.ProfileMeasurement; import org.apache.metron.profiler.ProfilePeriod; @@ -34,12 +39,6 @@ import org.apache.metron.profiler.hbase.RowKeyBuilder; import org.apache.metron.profiler.hbase.SaltyRowKeyBuilder; import org.apache.metron.profiler.hbase.ValueOnlyColumnBuilder; -import java.util.Collections; -import java.util.List; -import java.util.Random; -import java.util.concurrent.TimeUnit; -import java.util.function.Function; - /** * Writes ProfileMeasurement values that can be read during automated testing. */ @@ -50,11 +49,11 @@ public class ProfileWriter { private HBaseClient hbaseClient; private HBaseProfilerClient client; - public ProfileWriter(RowKeyBuilder rowKeyBuilder, ColumnBuilder columnBuilder, HTableInterface table, long periodDurationMillis) { + public ProfileWriter(RowKeyBuilder rowKeyBuilder, ColumnBuilder columnBuilder, TableProvider tableProvider, long periodDurationMillis, String tableName, Configuration configuration) { this.rowKeyBuilder = rowKeyBuilder; this.columnBuilder = columnBuilder; - this.hbaseClient = new HBaseClient((c, t) -> table, table.getConfiguration(), table.getName().getNameAsString()); - this.client = new HBaseProfilerClient(table, rowKeyBuilder, columnBuilder, periodDurationMillis); + this.hbaseClient = new HBaseClient(tableProvider, configuration, tableName); + this.client = new HBaseProfilerClient(tableProvider, rowKeyBuilder, columnBuilder, periodDurationMillis, tableName, configuration); } /** @@ -110,7 +109,7 @@ public class ProfileWriter { config.set("hbase.zookeeper.quorum", "node1"); HTableProvider provider = new HTableProvider(); - HTableInterface table = provider.getTable(config, "profiler"); + String tableName = "profiler"; long periodDurationMillis = TimeUnit.MINUTES.toMillis(15); long when = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2); @@ -119,7 +118,7 @@ public class ProfileWriter { .withEntity("192.168.66.121") .withPeriod(when, periodDurationMillis, TimeUnit.MILLISECONDS); - ProfileWriter writer = new ProfileWriter(rowKeyBuilder, columnBuilder, table, periodDurationMillis); + ProfileWriter writer = new ProfileWriter(rowKeyBuilder, columnBuilder, provider, periodDurationMillis, tableName, config); writer.write(measure, 2 * 24 * 4, Collections.emptyList(), val -> new Random().nextInt(10)); } } diff --git a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/GetProfileTest.java b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/GetProfileTest.java index 3fbed1c..c9f7d21 100644 --- a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/GetProfileTest.java +++ b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/GetProfileTest.java @@ -20,7 +20,21 @@ package org.apache.metron.profiler.client.stellar; -import org.apache.hadoop.hbase.client.HTableInterface; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.hbase.mock.MockHBaseTableProvider; import org.apache.metron.profiler.ProfileMeasurement; import org.apache.metron.profiler.client.ProfileWriter; @@ -38,21 +52,6 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; - -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; - /** * Tests the GetProfile class. */ @@ -89,13 +88,13 @@ public class GetProfileTest { @Before public void setup() { state = new HashMap<>(); - final HTableInterface table = MockHBaseTableProvider.addToCache(tableName, columnFamily); + final Table table = MockHBaseTableProvider.addToCache(tableName, columnFamily); // used to write values to be read during testing long periodDurationMillis = TimeUnit.MINUTES.toMillis(15); RowKeyBuilder rowKeyBuilder = new SaltyRowKeyBuilder(); ColumnBuilder columnBuilder = new ValueOnlyColumnBuilder(columnFamily); - profileWriter = new ProfileWriter(rowKeyBuilder, columnBuilder, table, periodDurationMillis); + profileWriter = new ProfileWriter(rowKeyBuilder, columnBuilder, new MockHBaseTableProvider(), periodDurationMillis, tableName, null); // global properties Map<String, Object> global = new HashMap<String, Object>() {{ diff --git a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/VerboseProfileTest.java b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/VerboseProfileTest.java index bd39007..ee02bd5 100644 --- a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/VerboseProfileTest.java +++ b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/VerboseProfileTest.java @@ -20,7 +20,28 @@ package org.apache.metron.profiler.client.stellar; -import org.apache.hadoop.hbase.client.HTableInterface; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS; +import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.ENTITY_KEY; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.GROUPS_KEY; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.PERIOD_END_KEY; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.PERIOD_KEY; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.PERIOD_START_KEY; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.PROFILE_KEY; +import static org.apache.metron.profiler.client.stellar.VerboseProfile.VALUE_KEY; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hbase.client.Table; +import org.apache.metron.hbase.TableProvider; import org.apache.metron.hbase.mock.MockHBaseTableProvider; import org.apache.metron.profiler.ProfileMeasurement; import org.apache.metron.profiler.client.ProfileWriter; @@ -36,27 +57,6 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; - -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS; -import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.ENTITY_KEY; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.GROUPS_KEY; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.PERIOD_END_KEY; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.PERIOD_KEY; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.PERIOD_START_KEY; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.PROFILE_KEY; -import static org.apache.metron.profiler.client.stellar.VerboseProfile.VALUE_KEY; - /** * Tests the VerboseProfile class. */ @@ -79,13 +79,15 @@ public class VerboseProfileTest { @Before public void setup() { state = new HashMap<>(); - final HTableInterface table = MockHBaseTableProvider.addToCache(tableName, columnFamily); + final Table table = MockHBaseTableProvider.addToCache(tableName, columnFamily); + TableProvider provider = new MockHBaseTableProvider(); // used to write values to be read during testing long periodDurationMillis = TimeUnit.MINUTES.toMillis(15); RowKeyBuilder rowKeyBuilder = new SaltyRowKeyBuilder(); ColumnBuilder columnBuilder = new ValueOnlyColumnBuilder(columnFamily); - profileWriter = new ProfileWriter(rowKeyBuilder, columnBuilder, table, periodDurationMillis); + profileWriter = new ProfileWriter(rowKeyBuilder, columnBuilder, provider, periodDurationMillis, + tableName, null); // global properties globals = new HashMap<String, Object>() {{ diff --git a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java index 148d970..7d728be 100644 --- a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java +++ b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java @@ -19,14 +19,13 @@ */ package org.apache.metron.profiler.spark; -import org.apache.hadoop.hbase.client.Durability; -import org.apache.metron.stellar.common.utils.ConversionUtils; +import static org.apache.metron.profiler.spark.reader.TelemetryReaders.JSON; import java.util.Map; import java.util.Properties; - -import static org.apache.metron.profiler.spark.reader.TelemetryReaders.JSON; -import static org.apache.metron.profiler.spark.reader.TelemetryReaders.TEXT; +import org.apache.hadoop.hbase.client.Durability; +import org.apache.metron.hbase.HTableProvider; +import org.apache.metron.stellar.common.utils.ConversionUtils; /** * Defines the configuration values recognized by the Batch Profiler. @@ -39,7 +38,7 @@ public enum BatchProfilerConfig { HBASE_SALT_DIVISOR("profiler.hbase.salt.divisor", 1000, Integer.class), - HBASE_TABLE_PROVIDER("profiler.hbase.table.provider", "org.apache.metron.hbase.HTableProvider", String.class), + HBASE_TABLE_PROVIDER("profiler.hbase.table.provider", HTableProvider.class.getName(), String.class), HBASE_TABLE_NAME("profiler.hbase.table", "profiler", String.class), diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java index 53b10f9..6af3b1d 100644 --- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java +++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java @@ -102,4 +102,4 @@ public class IndexConfig { throw new IllegalStateException("Unable to create index DAO: " + e.getMessage(), e); } } -} \ No newline at end of file +} diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/user/UserSettingsClient.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/user/UserSettingsClient.java index 7fa0f37..14a7ced 100644 --- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/user/UserSettingsClient.java +++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/user/UserSettingsClient.java @@ -28,7 +28,7 @@ import java.util.function.Supplier; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; @@ -45,7 +45,7 @@ public class UserSettingsClient { public static String USER_SETTINGS_HBASE_TABLE = "user.settings.hbase.table"; public static String USER_SETTINGS_HBASE_CF = "user.settings.hbase.cf"; - private HTableInterface userSettingsTable; + private Table userSettingsTable; private byte[] cf; private Supplier<Map<String, Object>> globalConfigSupplier; private TableProvider tableProvider; @@ -58,7 +58,7 @@ public class UserSettingsClient { this.tableProvider = tableProvider; } - public UserSettingsClient(HTableInterface userSettingsTable, byte[] cf) { + public UserSettingsClient(Table userSettingsTable, byte[] cf) { this.userSettingsTable = userSettingsTable; this.cf = cf; } @@ -84,7 +84,7 @@ public class UserSettingsClient { } } - public HTableInterface getTableInterface() { + public Table getTableInterface() { if(userSettingsTable == null) { init(globalConfigSupplier, tableProvider); } diff --git a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/HBaseConfigTest.java b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/HBaseConfigTest.java index 0d45f18..720f1ff 100644 --- a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/HBaseConfigTest.java +++ b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/HBaseConfigTest.java @@ -71,14 +71,6 @@ public class HBaseConfigTest { } @Test - public void hBaseClientShouldBeCreatedWithDefaultProvider() throws Exception { - when(globalConfigService.get()).thenReturn(new HashMap<String, Object>() {{ - put(EnrichmentConfigurations.TABLE_NAME, "enrichment_list_hbase_table_name"); - }}); - Assert.assertNotNull(hBaseConfig.hBaseClient()); - } - - @Test public void hBaseClientShouldBeCreatedWithSpecifiedProvider() throws Exception { when(globalConfigService.get()).thenReturn(new HashMap<String, Object>() {{ put(EnrichmentConfigurations.TABLE_PROVIDER, MockHBaseTableProvider.class.getName()); diff --git a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/TestConfig.java b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/TestConfig.java index c21068a..d363e91 100644 --- a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/TestConfig.java +++ b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/config/TestConfig.java @@ -27,7 +27,7 @@ import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.util.Bytes; import org.apache.kafka.clients.producer.KafkaProducer; @@ -196,7 +196,7 @@ public class TestConfig { public HBaseClient hBaseClient() throws RestException, IOException { final String cf = "t"; final String cq = "v"; - HTableInterface table = MockHBaseTableProvider.addToCache("enrichment_list", cf); + Table table = MockHBaseTableProvider.addToCache("enrichment_list", cf); List<String> enrichmentTypes = new ArrayList<String>() {{ add("foo"); add("bar"); diff --git a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/user/UserSettingsClientTest.java b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/user/UserSettingsClientTest.java index f856bcf..0fea0f4 100644 --- a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/user/UserSettingsClientTest.java +++ b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/user/UserSettingsClientTest.java @@ -18,28 +18,31 @@ package org.apache.metron.rest.user; +import static org.apache.metron.rest.user.UserSettingsClient.USER_SETTINGS_HBASE_CF; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.hadoop.hbase.client.*; -import org.apache.hadoop.hbase.util.Bytes; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.ExpectedException; - import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.HashMap; import java.util.Map; import java.util.Optional; import java.util.function.Supplier; - -import static org.apache.metron.rest.user.UserSettingsClient.USER_SETTINGS_HBASE_CF; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.mockito.Matchers.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.ResultScanner; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; public class UserSettingsClientTest { @@ -49,14 +52,14 @@ public class UserSettingsClientTest { private static ThreadLocal<ObjectMapper> _mapper = ThreadLocal.withInitial(() -> new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL)); - private HTableInterface userSettingsTable; + private Table userSettingsTable; private Supplier<Map<String, Object>> globalConfigSupplier; private UserSettingsClient userSettingsClient; private static byte[] cf = Bytes.toBytes("cf"); @Before public void setUp() throws Exception { - userSettingsTable = mock(HTableInterface.class); + userSettingsTable = mock(Table.class); globalConfigSupplier = () -> new HashMap<String, Object>() {{ put(USER_SETTINGS_HBASE_CF, "cf"); }}; diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/hbase/mr/PrunerMapper.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/hbase/mr/PrunerMapper.java index 82b4d3a..5c6d812 100644 --- a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/hbase/mr/PrunerMapper.java +++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/hbase/mr/PrunerMapper.java @@ -25,6 +25,7 @@ import org.apache.metron.enrichment.lookup.accesstracker.AccessTracker; import org.apache.metron.enrichment.lookup.accesstracker.AccessTrackerUtil; import java.io.IOException; +import org.apache.metron.hbase.HTableProvider; public class PrunerMapper extends TableMapper<ImmutableBytesWritable, Delete> { public static final String ACCESS_TRACKER_TABLE_CONF = "access_tracker_table"; @@ -38,7 +39,7 @@ public class PrunerMapper extends TableMapper<ImmutableBytesWritable, Delete> { String atTable = context.getConfiguration().get(ACCESS_TRACKER_TABLE_CONF); String atCF = context.getConfiguration().get(ACCESS_TRACKER_CF_CONF); String atName = context.getConfiguration().get(ACCESS_TRACKER_NAME_CONF); - HTable table = new HTable(context.getConfiguration(), atTable); + Table table = new HTableProvider().getTable(context.getConfiguration(), atTable); long timestamp = context.getConfiguration().getLong(TIMESTAMP_CONF, -1); if(timestamp < 0) { throw new IllegalStateException("Must specify a timestamp that is positive."); diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/HBaseExtractorState.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/HBaseExtractorState.java index f0ee3ad..fbf3bf3 100644 --- a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/HBaseExtractorState.java +++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/HBaseExtractorState.java @@ -19,20 +19,20 @@ package org.apache.metron.dataloads.nonbulk.flatfile; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.dataloads.extractor.Extractor; import org.apache.metron.enrichment.converter.HbaseConverter; import java.io.IOException; public class HBaseExtractorState { - private HTableInterface table; + private Table table; private Extractor extractor; private HbaseConverter converter; private FileSystem fs; private String cf; - public HBaseExtractorState(HTableInterface table, String cf, Extractor extractor, HbaseConverter converter, Configuration config) { + public HBaseExtractorState(Table table, String cf, Extractor extractor, HbaseConverter converter, Configuration config) { this.table = table; this.extractor = extractor; this.converter = converter; @@ -48,7 +48,7 @@ public class HBaseExtractorState { return cf; } - public HTableInterface getTable() { + public Table getTable() { return table; } diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/importer/LocalImporter.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/importer/LocalImporter.java index ec37585..87e60bc 100644 --- a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/importer/LocalImporter.java +++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/importer/LocalImporter.java @@ -17,9 +17,14 @@ */ package org.apache.metron.dataloads.nonbulk.flatfile.importer; +import java.io.IOException; +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; +import java.util.Optional; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.dataloads.extractor.Extractor; import org.apache.metron.dataloads.extractor.ExtractorHandler; import org.apache.metron.dataloads.nonbulk.flatfile.HBaseExtractorState; @@ -29,9 +34,6 @@ import org.apache.metron.enrichment.converter.HbaseConverter; import org.apache.metron.enrichment.lookup.LookupKV; import org.apache.metron.hbase.HTableProvider; -import java.io.*; -import java.util.*; - public class LocalImporter extends AbstractLocalImporter<LoadOptions, HBaseExtractorState> { public interface HTableProviderRetriever { @@ -87,7 +89,7 @@ public class LocalImporter extends AbstractLocalImporter<LoadOptions, HBaseExtra protected HBaseExtractorState initialValue() { try { String cf = (String) config.get(LoadOptions.HBASE_CF).get(); - HTableInterface table = provider.retrieve().getTable(hadoopConfig, (String) config.get(LoadOptions.HBASE_TABLE).get()); + Table table = provider.retrieve().getTable(hadoopConfig, (String) config.get(LoadOptions.HBASE_TABLE).get()); return new HBaseExtractorState(table, cf, handler.getExtractor(), new EnrichmentConverter(), hadoopConfig); } catch (IOException e1) { throw new IllegalStateException("Unable to get table: " + e1); diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java index de795c5..1cb20e0 100644 --- a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java +++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java @@ -44,9 +44,8 @@ import javax.xml.transform.TransformerFactory; import javax.xml.transform.dom.DOMSource; import javax.xml.transform.stream.StreamResult; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HTable; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Table; import org.apache.http.HttpHost; import org.apache.http.auth.AuthScope; import org.apache.http.auth.UsernamePasswordCredentials; @@ -72,6 +71,7 @@ import org.apache.metron.enrichment.converter.EnrichmentConverter; import org.apache.metron.enrichment.converter.EnrichmentKey; import org.apache.metron.enrichment.converter.EnrichmentValue; import org.apache.metron.enrichment.lookup.LookupKV; +import org.apache.metron.hbase.HTableProvider; import org.mitre.taxii.client.HttpClient; import org.mitre.taxii.messages.xml11.AnyMixedContentType; import org.mitre.taxii.messages.xml11.CollectionInformationRequest; @@ -116,7 +116,7 @@ public class TaxiiHandler extends TimerTask { private Extractor extractor; private String hbaseTable; private String columnFamily; - private Map<String, HTableInterface> connectionCache = new HashMap<>(); + private Map<String, Table> connectionCache = new HashMap<>(); private HttpClientContext context; private String collection; private String subscriptionId; @@ -147,8 +147,8 @@ public class TaxiiHandler extends TimerTask { LOG.info("Configured, starting polling {} for {}", endpoint, collection); } - protected synchronized HTableInterface getTable(String table) throws IOException { - HTableInterface ret = connectionCache.get(table); + protected synchronized Table getTable(String table) throws IOException { + Table ret = connectionCache.get(table); if(ret == null) { ret = createHTable(table); connectionCache.put(table, ret); @@ -156,8 +156,8 @@ public class TaxiiHandler extends TimerTask { return ret; } - protected synchronized HTableInterface createHTable(String tableInfo) throws IOException { - return new HTable(config, tableInfo); + protected synchronized Table createHTable(String tableInfo) throws IOException { + return new HTableProvider().getTable(config, tableInfo); } /** * The action to be performed by this timer task. @@ -222,7 +222,7 @@ public class TaxiiHandler extends TimerTask { kv.getValue().getMetadata().put("taxii_url", endpoint.toString()); kv.getValue().getMetadata().put("taxii_collection", collection); Put p = converter.toPut(columnFamily, kv.getKey(), kv.getValue()); - HTableInterface table = getTable(hbaseTable); + Table table = getTable(hbaseTable); table.put(p); LOG.info("Found Threat Intel: {} => ", kv.getKey(), kv.getValue()); } diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/hbase/mr/LeastRecentlyUsedPrunerIntegrationTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/hbase/mr/LeastRecentlyUsedPrunerIntegrationTest.java index d82be9d..947a5f4 100644 --- a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/hbase/mr/LeastRecentlyUsedPrunerIntegrationTest.java +++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/hbase/mr/LeastRecentlyUsedPrunerIntegrationTest.java @@ -18,12 +18,17 @@ package org.apache.metron.dataloads.hbase.mr; import com.google.common.collect.Iterables; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.logging.Level; import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.PosixParser; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseTestingUtility; -import org.apache.hadoop.hbase.client.HTable; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.util.GenericOptionsParser; @@ -36,21 +41,18 @@ import org.apache.metron.enrichment.lookup.LookupKey; import org.apache.metron.enrichment.lookup.accesstracker.BloomAccessTracker; import org.apache.metron.enrichment.lookup.accesstracker.PersistentAccessTracker; import org.apache.metron.test.utils.UnitTestHelper; -import org.junit.*; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.logging.Level; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; public class LeastRecentlyUsedPrunerIntegrationTest { /** The test util. */ private static HBaseTestingUtility testUtil; /** The test table. */ - private static HTable testTable; - private static HTable atTable; + private static Table testTable; + private static Table atTable; private static final String tableName = "malicious_domains"; private static final String cf = "cf"; private static final String atTableName = "access_trackers"; diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoaderIntegrationTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoaderIntegrationTest.java index a1d98f6..274989a 100644 --- a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoaderIntegrationTest.java +++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoaderIntegrationTest.java @@ -17,8 +17,23 @@ */ package org.apache.metron.dataloads.nonbulk.flatfile; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.startsWith; + +import java.io.File; +import java.io.FileOutputStream; import java.io.OutputStreamWriter; +import java.io.PrintWriter; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.logging.Level; +import java.util.zip.GZIPOutputStream; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; import org.adrianwalker.multilinestring.Multiline; import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.PosixParser; @@ -29,9 +44,9 @@ import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.client.Delete; -import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.util.GenericOptionsParser; import org.apache.metron.common.configuration.ConfigurationsUtils; @@ -47,28 +62,12 @@ import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; -import java.io.File; -import java.io.FileOutputStream; -import java.io.PrintWriter; -import java.nio.file.Files; -import java.nio.file.StandardOpenOption; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.logging.Level; -import java.util.zip.GZIPOutputStream; -import java.util.zip.ZipEntry; -import java.util.zip.ZipOutputStream; - -import static org.hamcrest.CoreMatchers.equalTo; -import static org.hamcrest.CoreMatchers.startsWith; - public class SimpleEnrichmentFlatFileLoaderIntegrationTest { private static HBaseTestingUtility testUtil; /** The test table. */ - private static HTable testTable; + private static Table testTable; private static Configuration config = null; private static TestingServer testZkServer; private static String zookeeperUrl; diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiIntegrationTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiIntegrationTest.java index a54c21b..6cb94d3 100644 --- a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiIntegrationTest.java +++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiIntegrationTest.java @@ -24,7 +24,7 @@ import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.PosixParser; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.util.GenericOptionsParser; import org.apache.metron.dataloads.extractor.Extractor; @@ -99,7 +99,7 @@ public class TaxiiIntegrationTest { Extractor extractor = new TransformFilterExtractorDecorator(new StixExtractor()); TaxiiHandler handler = new TaxiiHandler(TaxiiConnectionConfig.load(taxiiConnectionConfig), extractor, config ) { @Override - protected synchronized HTableInterface createHTable(String tableInfo) throws IOException { + protected synchronized Table createHTable(String tableInfo) throws IOException { return provider.addToCache("threat_intel", "cf"); } }; diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java index 913d89c..ec325e2 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/cif/CIFHbaseAdapter.java @@ -27,11 +27,12 @@ import java.util.Map; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HConnection; -import org.apache.hadoop.hbase.client.HConnectionManager; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.enrichment.cache.CacheKey; import org.apache.metron.enrichment.interfaces.EnrichmentAdapter; import org.json.simple.JSONObject; @@ -43,7 +44,7 @@ public class CIFHbaseAdapter implements EnrichmentAdapter<CacheKey>,Serializable private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); private static final long serialVersionUID = 1L; private String _tableName; - private HTableInterface table; + private Table table; private String _quorum; private String _port; @@ -103,8 +104,8 @@ public class CIFHbaseAdapter implements EnrichmentAdapter<CacheKey>,Serializable try { LOGGER.debug("=======Connecting to HBASE==========="); LOGGER.debug("=======ZOOKEEPER = {}", conf.get("hbase.zookeeper.quorum")); - HConnection connection = HConnectionManager.createConnection(conf); - table = connection.getTable(_tableName); + Connection connection = ConnectionFactory.createConnection(conf); + table = connection.getTable(TableName.valueOf(_tableName)); return true; } catch (IOException e) { LOGGER.debug("=======Unable to Connect to HBASE==========="); diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java index 06f9d21..d1b0a31 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/simplehbase/SimpleHBaseAdapter.java @@ -27,6 +27,7 @@ import java.util.List; import java.util.Map; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.client.Connection; import org.apache.metron.common.utils.LazyLogger; import org.apache.metron.common.utils.LazyLoggerFactory; import org.apache.metron.enrichment.cache.CacheKey; @@ -43,6 +44,7 @@ public class SimpleHBaseAdapter implements EnrichmentAdapter<CacheKey>,Serializa protected static final LazyLogger LOG = LazyLoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); protected SimpleHBaseConfig config; protected EnrichmentLookup lookup; + protected Connection connection; public SimpleHBaseAdapter() { } diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java index 8515c68..9adcb48 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapter.java @@ -26,6 +26,7 @@ import java.util.Map; import java.util.UUID; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.client.Connection; import org.apache.metron.enrichment.cache.CacheKey; import org.apache.metron.enrichment.converter.EnrichmentKey; import org.apache.metron.enrichment.interfaces.EnrichmentAdapter; @@ -41,6 +42,7 @@ public class ThreatIntelAdapter implements EnrichmentAdapter<CacheKey>,Serializa protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); protected ThreatIntelConfig config; protected EnrichmentLookup lookup; + protected Connection connection; public ThreatIntelAdapter() { } diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/converter/EnrichmentHelper.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/converter/EnrichmentHelper.java index 475ee8c..b8198df 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/converter/EnrichmentHelper.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/converter/EnrichmentHelper.java @@ -17,17 +17,16 @@ */ package org.apache.metron.enrichment.converter; -import org.apache.hadoop.hbase.client.HTableInterface; +import java.io.IOException; import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.enrichment.lookup.LookupKV; -import java.io.IOException; - public enum EnrichmentHelper { INSTANCE; EnrichmentConverter converter = new EnrichmentConverter(); - public void load(HTableInterface table, String cf, Iterable<LookupKV<EnrichmentKey, EnrichmentValue>> results) throws IOException { + public void load(Table table, String cf, Iterable<LookupKV<EnrichmentKey, EnrichmentValue>> results) throws IOException { for(LookupKV<EnrichmentKey, EnrichmentValue> result : results) { Put put = converter.toPut(cf, result.getKey(), result.getValue()); table.put(put); diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/EnrichmentLookup.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/EnrichmentLookup.java index 3ee9754..c148e9e 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/EnrichmentLookup.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/EnrichmentLookup.java @@ -19,7 +19,7 @@ package org.apache.metron.enrichment.lookup; import com.google.common.collect.Iterables; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.Result; import org.apache.metron.enrichment.converter.HbaseConverter; import org.apache.metron.enrichment.converter.EnrichmentConverter; @@ -38,14 +38,14 @@ import java.util.List; public class EnrichmentLookup extends Lookup<EnrichmentLookup.HBaseContext, EnrichmentKey, LookupKV<EnrichmentKey,EnrichmentValue>> implements AutoCloseable { public static class HBaseContext { - private HTableInterface table; + private Table table; private String columnFamily; - public HBaseContext(HTableInterface table, String columnFamily) { + public HBaseContext(Table table, String columnFamily) { this.table = table; this.columnFamily = columnFamily; } - public HTableInterface getTable() { return table; } + public Table getTable() { return table; } public String getColumnFamily() { return columnFamily; } } @@ -84,7 +84,7 @@ public class EnrichmentLookup extends Lookup<EnrichmentLookup.HBaseContext, Enri if(Iterables.isEmpty(key)) { return Collections.emptyList(); } - HTableInterface table = Iterables.getFirst(key, null).getContext().getTable(); + Table table = Iterables.getFirst(key, null).getContext().getTable(); for(boolean b : table.existsAll(keysToGets(key))) { ret.add(b); } @@ -99,7 +99,7 @@ public class EnrichmentLookup extends Lookup<EnrichmentLookup.HBaseContext, Enri if(Iterables.isEmpty(keys)) { return Collections.emptyList(); } - HTableInterface table = Iterables.getFirst(keys, null).getContext().getTable(); + Table table = Iterables.getFirst(keys, null).getContext().getTable(); List<LookupKV<EnrichmentKey, EnrichmentValue>> ret = new ArrayList<>(); Iterator<KeyWithContext<EnrichmentKey, HBaseContext>> keyWithContextIterator = keys.iterator(); for(Result result : table.get(keysToGets(keys))) { @@ -115,14 +115,14 @@ public class EnrichmentLookup extends Lookup<EnrichmentLookup.HBaseContext, Enri } } - private HTableInterface table; - public EnrichmentLookup(HTableInterface table, String columnFamily, AccessTracker tracker) { + private Table table; + public EnrichmentLookup(Table table, String columnFamily, AccessTracker tracker) { this.table = table; this.setLookupHandler(new Handler(columnFamily)); this.setAccessTracker(tracker); } - public HTableInterface getTable() { + public Table getTable() { return table; } diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/AccessTrackerUtil.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/AccessTrackerUtil.java index 5d880f2..add1f0f 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/AccessTrackerUtil.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/AccessTrackerUtil.java @@ -44,13 +44,13 @@ public enum AccessTrackerUtil { } - public void persistTracker(HTableInterface accessTrackerTable, String columnFamily, PersistentAccessTracker.AccessTrackerKey key, AccessTracker underlyingTracker) throws IOException { + public void persistTracker(Table accessTrackerTable, String columnFamily, PersistentAccessTracker.AccessTrackerKey key, AccessTracker underlyingTracker) throws IOException { Put put = new Put(key.toRowKey()); put.add(Bytes.toBytes(columnFamily), COLUMN, serializeTracker(underlyingTracker)); accessTrackerTable.put(put); } - public Iterable<AccessTracker> loadAll(HTableInterface accessTrackerTable, final String columnFamily, final String name, final long earliest) throws IOException { + public Iterable<AccessTracker> loadAll(Table accessTrackerTable, final String columnFamily, final String name, final long earliest) throws IOException { Scan scan = new Scan(PersistentAccessTracker.AccessTrackerKey.getTimestampScanKey(name, earliest)); ResultScanner scanner = accessTrackerTable.getScanner(scan); return Iterables.transform(scanner, new Function<Result, AccessTracker>() { diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentAccessTracker.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentAccessTracker.java index fd0cd61..2c63ea7 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentAccessTracker.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentAccessTracker.java @@ -26,7 +26,7 @@ import java.lang.invoke.MethodHandles; import java.util.Map; import java.util.Timer; import java.util.TimerTask; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.enrichment.lookup.LookupKey; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,7 +35,7 @@ public class PersistentAccessTracker implements AccessTracker { private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); private static final long serialVersionUID = 1L; - public static class AccessTrackerKey { + public static class AccessTrackerKey { String name; String containerName; long timestamp; @@ -102,7 +102,7 @@ public class PersistentAccessTracker implements AccessTracker { } final Object sync = new Object(); - HTableInterface accessTrackerTable; + Table accessTrackerTable; String accessTrackerColumnFamily; AccessTracker underlyingTracker; long timestamp = System.currentTimeMillis(); @@ -113,7 +113,7 @@ public class PersistentAccessTracker implements AccessTracker { public PersistentAccessTracker( String name , String containerName - , HTableInterface accessTrackerTable + , Table accessTrackerTable , String columnFamily , AccessTracker underlyingTracker , long maxMillisecondsBetweenPersists diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentBloomTrackerCreator.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentBloomTrackerCreator.java index a7b1b5f..0cc3d30 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentBloomTrackerCreator.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/lookup/accesstracker/PersistentBloomTrackerCreator.java @@ -17,14 +17,13 @@ */ package org.apache.metron.enrichment.lookup.accesstracker; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.metron.stellar.common.utils.ConversionUtils; -import org.apache.metron.hbase.TableProvider; - import java.io.IOException; import java.util.Map; import java.util.UUID; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.metron.hbase.TableProvider; +import org.apache.metron.stellar.common.utils.ConversionUtils; public class PersistentBloomTrackerCreator implements AccessTrackerCreator { diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/stellar/SimpleHBaseEnrichmentFunctions.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/stellar/SimpleHBaseEnrichmentFunctions.java index f8cdf36..492b391 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/stellar/SimpleHBaseEnrichmentFunctions.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/stellar/SimpleHBaseEnrichmentFunctions.java @@ -26,7 +26,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.enrichment.converter.EnrichmentKey; import org.apache.metron.enrichment.converter.EnrichmentValue; import org.apache.metron.enrichment.lookup.EnrichmentLookup; @@ -50,11 +50,11 @@ public class SimpleHBaseEnrichmentFunctions { private static TableProvider provider; - private static class Table { + private static class WrapperTable { String name; String columnFamily; - public Table(String name, String columnFamily) { + public WrapperTable(String name, String columnFamily) { this.name = name; this.columnFamily = columnFamily; } @@ -72,7 +72,7 @@ public class SimpleHBaseEnrichmentFunctions { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; - Table table = (Table) o; + WrapperTable table = (WrapperTable) o; if (name != null ? !name.equals(table.name) : table.name != null) return false; return columnFamily != null ? columnFamily.equals(table.columnFamily) : table.columnFamily == null; @@ -133,7 +133,7 @@ public class SimpleHBaseEnrichmentFunctions { ) public static class EnrichmentExists implements StellarFunction { boolean initialized = false; - private static Cache<Table, EnrichmentLookup> enrichmentCollateralCache = CacheBuilder.newBuilder() + private static Cache<WrapperTable, EnrichmentLookup> enrichmentCollateralCache = CacheBuilder.newBuilder() .build(); @Override public Object apply(List<Object> args, Context context) throws ParseException { @@ -151,11 +151,11 @@ public class SimpleHBaseEnrichmentFunctions { if(enrichmentType == null || indicator == null) { return false; } - final Table key = new Table(table, cf); + final WrapperTable key = new WrapperTable(table, cf); EnrichmentLookup lookup = null; try { lookup = enrichmentCollateralCache.get(key, () -> { - HTableInterface hTable = provider.getTable(HBaseConfiguration.create(), key.name); + Table hTable = provider.getTable(HBaseConfiguration.create(), key.name); return new EnrichmentLookup(hTable, key.columnFamily, tracker); } ); @@ -210,7 +210,7 @@ public class SimpleHBaseEnrichmentFunctions { ) public static class EnrichmentGet implements StellarFunction { boolean initialized = false; - private static Cache<Table, EnrichmentLookup> enrichmentCollateralCache = CacheBuilder.newBuilder() + private static Cache<WrapperTable, EnrichmentLookup> enrichmentCollateralCache = CacheBuilder.newBuilder() .build(); @Override public Object apply(List<Object> args, Context context) throws ParseException { @@ -228,11 +228,11 @@ public class SimpleHBaseEnrichmentFunctions { if(enrichmentType == null || indicator == null) { return new HashMap<String, Object>(); } - final Table key = new Table(table, cf); + final WrapperTable key = new WrapperTable(table, cf); EnrichmentLookup lookup = null; try { lookup = enrichmentCollateralCache.get(key, () -> { - HTableInterface hTable = provider.getTable(HBaseConfiguration.create(), key.name); + Table hTable = provider.getTable(HBaseConfiguration.create(), key.name); return new EnrichmentLookup(hTable, key.columnFamily, tracker); } ); diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/utils/EnrichmentUtils.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/utils/EnrichmentUtils.java index 9a36a87..cec9958 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/utils/EnrichmentUtils.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/main/java/org/apache/metron/enrichment/utils/EnrichmentUtils.java @@ -26,7 +26,7 @@ import java.util.HashMap; import java.util.Map; import javax.annotation.Nullable; import org.apache.commons.lang.StringUtils; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.metron.common.configuration.enrichment.EnrichmentConfig; import org.apache.metron.enrichment.converter.EnrichmentKey; import org.apache.metron.enrichment.lookup.EnrichmentLookup; @@ -45,8 +45,8 @@ public class EnrichmentUtils { public static class TypeToKey implements Function<String, KeyWithContext<EnrichmentKey, EnrichmentLookup.HBaseContext>> { private final String indicator; private final EnrichmentConfig config; - private final HTableInterface table; - public TypeToKey(String indicator, HTableInterface table, EnrichmentConfig config) { + private final Table table; + public TypeToKey(String indicator, Table table, EnrichmentConfig config) { this.indicator = indicator; this.config = config; this.table = table; diff --git a/metron-platform/metron-enrichment/metron-enrichment-common/src/test/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapterTest.java b/metron-platform/metron-enrichment/metron-enrichment-common/src/test/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapterTest.java index bf51b4c..3569c6a 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-common/src/test/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapterTest.java +++ b/metron-platform/metron-enrichment/metron-enrichment-common/src/test/java/org/apache/metron/enrichment/adapters/threatintel/ThreatIntelAdapterTest.java @@ -19,7 +19,7 @@ package org.apache.metron.enrichment.adapters.threatintel; import org.adrianwalker.multilinestring.Multiline; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.log4j.Level; import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig; import org.apache.metron.enrichment.cache.CacheKey; @@ -53,7 +53,7 @@ public class ThreatIntelAdapterTest { public ExceptionProvider() {}; @Override - public HTableInterface getTable(Configuration config, String tableName) throws IOException { + public Table getTable(Configuration config, String tableName) throws IOException { throw new IOException(); } } diff --git a/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorIntegrationTest.java b/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorIntegrationTest.java index 1ff83de..2bc756e 100644 --- a/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorIntegrationTest.java +++ b/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorIntegrationTest.java @@ -37,13 +37,14 @@ import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.Delete; -import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; import org.apache.log4j.Level; import org.apache.metron.common.configuration.ConfigurationsUtils; import org.apache.metron.dataloads.hbase.mr.HBaseUtil; import org.apache.metron.enrichment.converter.EnrichmentKey; +import org.apache.metron.hbase.HTableProvider; import org.apache.metron.hbase.helper.HelperDao; import org.apache.metron.integration.BaseIntegrationTest; import org.apache.metron.integration.ComponentRunner; @@ -65,8 +66,8 @@ public class EnrichmentCoprocessorIntegrationTest extends BaseIntegrationTest { private static ZKServerComponent zookeeperComponent; private static ComponentRunner componentRunner; private static HBaseTestingUtility testUtil; - private static HTable enrichmentTable; - private static HTable enrichmentListTable; + private static Table enrichmentTable; + private static Table enrichmentListTable; private static Configuration hBaseConfig; /* @@ -75,7 +76,7 @@ public class EnrichmentCoprocessorIntegrationTest extends BaseIntegrationTest { /** * { - * "enrichment.list.hbase.provider.impl" : "org.apache.metron.hbase.HTableProvider", + * "enrichment.list.hbase.provider.impl" : "%PROVIDER_NAME%", * "enrichment.list.hbase.table" : "%TABLE_NAME%", * "enrichment.list.hbase.cf" : "%COLUMN_FAMILY%" * } @@ -89,7 +90,7 @@ public class EnrichmentCoprocessorIntegrationTest extends BaseIntegrationTest { // don't need the properties for anything else now, but could extract var if desired. startZookeeper(new Properties()); globalConfig = globalConfig.replace("%TABLE_NAME%", ENRICHMENT_LIST_TABLE) - .replace("%COLUMN_FAMILY%", COLUMN_FAMILY); + .replace("%COLUMN_FAMILY%", COLUMN_FAMILY).replace("%PROVIDER_NAME%", HTableProvider.class.getName()); uploadGlobalConfigToZK(globalConfig); configureAndStartHBase(); addCoprocessor(enrichmentTable.getName()); diff --git a/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorTest.java b/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorTest.java index fdf7493..cf9f32a 100644 --- a/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorTest.java +++ b/metron-platform/metron-hbase-server/src/test/java/org/apache/metron/hbase/coprocessor/EnrichmentCoprocessorTest.java @@ -37,8 +37,8 @@ import java.util.List; import java.util.Map; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.coprocessor.ObserverContext; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; import org.apache.metron.common.configuration.EnrichmentConfigurations; @@ -121,9 +121,10 @@ public class EnrichmentCoprocessorTest { } @Override - public HTableInterface getTable(Configuration config, String tableName) throws IOException { + public Table getTable(Configuration config, String tableName) throws IOException { return null; // not used for instantiation test } + } @Test diff --git a/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/HTableProvider.java b/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/HTableProvider.java index a323beb..201d067 100644 --- a/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/HTableProvider.java +++ b/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/HTableProvider.java @@ -18,13 +18,49 @@ package org.apache.metron.hbase; import java.io.IOException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HTable; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.client.Table; public class HTableProvider implements TableProvider { - @Override - public HTableInterface getTable(Configuration config, String tableName) throws IOException { - return new HTable(config, tableName); + + private static class RetryingConnection { + + private Configuration config; + private Connection conn; + + RetryingConnection(Configuration config) { + this.config = config; + } + + public Connection getUnderlying() throws IOException { + if (conn == null || conn.isClosed()) { + conn = ConnectionFactory.createConnection(config); + } + return conn; } + } + + /** + * We have to handle serialization issues with Storm via indirections. Rather than re-implement + * the interface everywhere we touch HBase, we can use a lazy initialization scheme to encapsulate + * this within the HTableProvider. This is a sort of poor man's connection pool. + */ + private static Map<Configuration, ThreadLocal<RetryingConnection>> connMap = new ConcurrentHashMap<>(); + + @Override + public Table getTable(Configuration config, String tableName) + throws IOException { + return getConnection(config).getTable(TableName.valueOf(tableName)); + } + + private Connection getConnection(Configuration config) throws IOException { + ThreadLocal<RetryingConnection> threadLocal = connMap.computeIfAbsent(config, c -> ThreadLocal.withInitial(() -> new RetryingConnection(config))); + return threadLocal.get().getUnderlying(); + } + } diff --git a/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/TableProvider.java b/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/TableProvider.java index 8511f75..f5b68d6 100644 --- a/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/TableProvider.java +++ b/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/TableProvider.java @@ -17,16 +17,16 @@ */ package org.apache.metron.hbase; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HTableInterface; - import java.io.IOException; import java.io.Serializable; import java.lang.reflect.InvocationTargetException; import java.util.function.Supplier; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.client.Table; public interface TableProvider extends Serializable { - HTableInterface getTable(Configuration config, String tableName) throws IOException; + + Table getTable(Configuration config, String tableName) throws IOException; /** * Factory method that creates TableProviders. diff --git a/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/client/HBaseClient.java b/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/client/HBaseClient.java index d0d934e..3fe1cfa 100644 --- a/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/client/HBaseClient.java +++ b/metron-platform/metron-hbase/metron-hbase-common/src/main/java/org/apache/metron/hbase/client/HBaseClient.java @@ -30,17 +30,17 @@ import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.client.Durability; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Increment; import org.apache.hadoop.hbase.client.Mutation; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.metron.hbase.TableProvider; import org.apache.metron.hbase.ColumnList; import org.apache.metron.hbase.HBaseProjectionCriteria; +import org.apache.metron.hbase.TableProvider; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -64,8 +64,11 @@ public class HBaseClient implements Closeable { /** * The HBase table this client interacts with. */ - private HTableInterface table; + private Table table; + /** + * New hbase client. + */ public HBaseClient(TableProvider provider, final Configuration configuration, final String tableName) { this.mutations = new ArrayList<>(); this.gets = new ArrayList<>(); @@ -294,7 +297,7 @@ public class HBaseClient implements Closeable { * @param table The table to retrieve the name of. * @return The name of the table */ - private static String tableName(HTableInterface table) { + private static String tableName(Table table) { String tableName = "null"; if(table != null) { if(table.getName() != null) { diff --git a/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java b/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java index 1983fc7..cb58b57 100644 --- a/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java +++ b/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/client/HBaseClientTest.java @@ -20,6 +20,19 @@ package org.apache.metron.hbase.client; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyListOf; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseTestingUtility; @@ -27,13 +40,13 @@ import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Durability; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.metron.hbase.TableProvider; import org.apache.metron.hbase.ColumnList; import org.apache.metron.hbase.HBaseProjectionCriteria; +import org.apache.metron.hbase.TableProvider; import org.junit.After; import org.junit.AfterClass; import org.junit.Assert; @@ -41,20 +54,6 @@ import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.TimeUnit; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertTrue; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyListOf; -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - /** * Tests the HBaseClient */ @@ -64,7 +63,7 @@ public class HBaseClientTest { private static HBaseTestingUtility util; private static HBaseClient client; - private static HTableInterface table; + private static Table table; private static Admin admin; private static byte[] cf = Bytes.toBytes("cf"); private static byte[] column = Bytes.toBytes("column"); @@ -265,7 +264,7 @@ public class HBaseClientTest { @Test(expected = RuntimeException.class) public void testFailureToMutate() throws IOException, InterruptedException { // used to trigger a failure condition in `HbaseClient.mutate` - HTableInterface table = mock(HTableInterface.class); + Table table = mock(Table.class); doThrow(new IOException("exception!")).when(table).batch(any(), any()); TableProvider tableProvider = mock(TableProvider.class); @@ -279,7 +278,7 @@ public class HBaseClientTest { @Test(expected = RuntimeException.class) public void testFailureToGetAll() throws IOException { // used to trigger a failure condition in `HbaseClient.getAll` - HTableInterface table = mock(HTableInterface.class); + Table table = mock(Table.class); when(table.get(anyListOf(Get.class))).thenThrow(new IOException("exception!")); TableProvider tableProvider = mock(TableProvider.class); diff --git a/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHBaseTableProvider.java b/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHBaseTableProvider.java index 57981ac..3d4316d 100644 --- a/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHBaseTableProvider.java +++ b/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHBaseTableProvider.java @@ -17,27 +17,26 @@ */ package org.apache.metron.hbase.mock; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HTableInterface; -import org.apache.metron.hbase.TableProvider; - import java.io.IOException; import java.io.Serializable; import java.util.HashMap; import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.client.Table; +import org.apache.metron.hbase.TableProvider; public class MockHBaseTableProvider implements Serializable, TableProvider { - private static Map<String, HTableInterface> _cache = new HashMap<>(); - public HTableInterface getTable(Configuration config, String tableName) throws IOException { - HTableInterface ret = _cache.get(tableName); + private static Map<String, Table> _cache = new HashMap<>(); + public Table getTable(Configuration configuration, String tableName) throws IOException { + Table ret = _cache.get(tableName); return ret; } - public static HTableInterface getFromCache(String tableName) { + public static Table getFromCache(String tableName) { return _cache.get(tableName); } - public static HTableInterface addToCache(String tableName, String... columnFamilies) { + public static Table addToCache(String tableName, String... columnFamilies) { MockHTable ret = new MockHTable(tableName, columnFamilies); _cache.put(tableName, ret); return ret; @@ -46,4 +45,5 @@ public class MockHBaseTableProvider implements Serializable, TableProvider { public static void clear() { _cache.clear(); } + } diff --git a/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHTable.java b/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHTable.java index 738275d..521f055 100644 --- a/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHTable.java +++ b/metron-platform/metron-hbase/metron-hbase-common/src/test/java/org/apache/metron/hbase/mock/MockHTable.java @@ -23,7 +23,16 @@ import com.google.protobuf.Descriptors; import com.google.protobuf.Message; import com.google.protobuf.Service; import com.google.protobuf.ServiceException; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.NavigableSet; +import java.util.NoSuchElementException; +import java.util.TreeMap; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HColumnDescriptor; @@ -34,7 +43,6 @@ import org.apache.hadoop.hbase.client.Append; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Durability; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Increment; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; @@ -42,31 +50,19 @@ import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Row; import org.apache.hadoop.hbase.client.RowMutations; import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.coprocessor.Batch; import org.apache.hadoop.hbase.filter.CompareFilter; import org.apache.hadoop.hbase.filter.Filter; import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.metron.hbase.TableProvider; - -import java.io.IOException; -import java.io.Serializable; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.NavigableMap; -import java.util.NavigableSet; -import java.util.NoSuchElementException; -import java.util.TreeMap; /** * MockHTable. * * This implementation is a selected excerpt from https://gist.github.com/agaoglu/613217 */ -public class MockHTable implements HTableInterface { +public class MockHTable implements Table { private final String tableName; @@ -124,7 +120,6 @@ public class MockHTable implements HTableInterface { } } - @Override public byte[] getTableName() { return Bytes.toBytes(tableName); } @@ -191,7 +186,6 @@ public class MockHTable implements HTableInterface { return ret; } - @Override public Boolean[] exists(List<Get> list) throws IOException { Boolean[] ret = new Boolean[list.size()]; int i = 0; @@ -316,7 +310,6 @@ public class MockHTable implements HTableInterface { * @deprecated */ @Deprecated - @Override public Result getRowOrBefore(byte[] bytes, byte[] bytes1) throws IOException { throw new UnsupportedOperationException(); } @@ -608,17 +601,14 @@ public class MockHTable implements HTableInterface { * @deprecated */ @Deprecated - @Override public long incrementColumnValue(byte[] bytes, byte[] bytes1, byte[] bytes2, long l, boolean b) throws IOException { throw new UnsupportedOperationException(); } - @Override public boolean isAutoFlush() { return autoflush; } - @Override public void flushCommits() throws IOException { } @@ -650,17 +640,14 @@ public class MockHTable implements HTableInterface { * @deprecated */ @Deprecated - @Override public void setAutoFlush(boolean b) { autoflush = b; } - @Override public void setAutoFlush(boolean b, boolean b1) { autoflush = b; } - @Override public void setAutoFlushTo(boolean b) { autoflush = b; } diff --git a/metron-platform/metron-indexing/metron-indexing-common/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java b/metron-platform/metron-indexing/metron-indexing-common/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java index 6b246d4..723f0aa 100644 --- a/metron-platform/metron-indexing/metron-indexing-common/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java +++ b/metron-platform/metron-indexing/metron-indexing-common/src/main/java/org/apache/metron/indexing/dao/HBaseDao.java @@ -32,7 +32,7 @@ import java.util.Optional; import java.util.stream.Collectors; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.util.Bytes; @@ -63,7 +63,7 @@ import org.apache.metron.indexing.dao.update.Document; public class HBaseDao implements IndexDao { public static String HBASE_TABLE = "update.hbase.table"; public static String HBASE_CF = "update.hbase.cf"; - private HTableInterface tableInterface; + private Table tableInterface; private byte[] cf; private AccessConfig config; @@ -168,7 +168,7 @@ public class HBaseDao implements IndexDao { } } - public HTableInterface getTableInterface() { + public Table getTableInterface() { if(tableInterface == null) { init(config); } diff --git a/metron-platform/metron-writer/metron-writer-common/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java b/metron-platform/metron-writer/metron-writer-common/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java index dec6926..448dcfd 100644 --- a/metron-platform/metron-writer/metron-writer-common/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java +++ b/metron-platform/metron-writer/metron-writer-common/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java @@ -26,16 +26,16 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.Put; import org.apache.metron.common.configuration.writer.WriterConfiguration; import org.apache.metron.common.writer.BulkMessageWriter; +import org.apache.metron.hbase.HTableProvider; import org.apache.metron.stellar.common.utils.ConversionUtils; import org.apache.metron.common.utils.ReflectionUtils; import org.apache.metron.enrichment.converter.EnrichmentConverter; import org.apache.metron.enrichment.converter.EnrichmentKey; import org.apache.metron.enrichment.converter.EnrichmentValue; -import org.apache.metron.hbase.HTableProvider; import org.apache.metron.hbase.TableProvider; import org.apache.metron.writer.AbstractWriter; import org.apache.metron.common.writer.BulkWriterResponse; @@ -118,7 +118,7 @@ public class SimpleHbaseEnrichmentWriter extends AbstractWriter implements BulkM private transient EnrichmentConverter converter; private String tableName; private String cf; - private HTableInterface table; + private Table table; private TableProvider provider; private Map.Entry<Object, KeyTransformer> keyTransformer; @@ -191,7 +191,7 @@ public class SimpleHbaseEnrichmentWriter extends AbstractWriter implements BulkM return provider; } - public HTableInterface getTable(String tableName, String cf) throws IOException { + public Table getTable(String tableName, String cf) throws IOException { synchronized(this) { boolean isInitial = this.tableName == null || this.cf == null; boolean isValid = tableName != null && cf != null; @@ -213,7 +213,7 @@ public class SimpleHbaseEnrichmentWriter extends AbstractWriter implements BulkM } } - public HTableInterface getTable(Map<String, Object> config) throws IOException { + public Table getTable(Map<String, Object> config) throws IOException { return getTable(Configurations.HBASE_TABLE.getAndConvert(config, String.class) ,Configurations.HBASE_CF.getAndConvert(config, String.class) ); @@ -321,7 +321,7 @@ public class SimpleHbaseEnrichmentWriter extends AbstractWriter implements BulkM ) throws Exception { Map<String, Object> sensorConfig = configurations.getSensorConfig(sensorType); - HTableInterface table = getTable(sensorConfig); + Table table = getTable(sensorConfig); KeyTransformer transformer = getTransformer(sensorConfig); Object enrichmentTypeObj = Configurations.ENRICHMENT_TYPE.get(sensorConfig); String enrichmentType = enrichmentTypeObj == null?null:enrichmentTypeObj.toString();