@Test public void testParseColumns() throws IOException { runner.setProperty(GetHBase.COLUMNS, "cf1,cf2:cq1,cf3"); proc.parseColumns(runner.getProcessContext()); final List<Column> expectedCols = new ArrayList<>(); expectedCols.add(new Column("cf1".getBytes(Charset.forName("UTF-8")), null)); expectedCols.add(new Column("cf2".getBytes(Charset.forName("UTF-8")), "cq1".getBytes(Charset.forName("UTF-8")))); expectedCols.add(new Column("cf3".getBytes(Charset.forName("UTF-8")), null)); final List<Column> actualColumns = proc.getColumns(); Assert.assertNotNull(actualColumns); Assert.assertEquals(expectedCols.size(), actualColumns.size()); for (final Column expectedCol : expectedCols) { boolean found = false; for (final Column providedCol : actualColumns) { if (expectedCol.equals(providedCol)) { found = true; break; } } Assert.assertTrue("Didn't find expected column", found); } }
@OnScheduled public void parseColumns(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() < 0) { final DistributedMapCacheClient client = context.getProperty(DISTRIBUTED_CACHE_SERVICE).asControllerService(DistributedMapCacheClient.class); final ScanResult scanResult = getState(client); if (scanResult != null) { storeState(scanResult, context.getStateManager()); } clearState(client); } final String columnsValue = context.getProperty(COLUMNS).getValue(); final String[] columns = (columnsValue == null || columnsValue.isEmpty() ? new String[0] : columnsValue.split(",")); this.columns.clear(); for (final String column : columns) { if (column.contains(":")) { final String[] parts = column.split(":"); final byte[] cf = parts[0].getBytes(Charset.forName("UTF-8")); final byte[] cq = parts[1].getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, cq)); } else { final byte[] cf = column.getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, null)); } } }
GetHBase extends AbstractProcessor { @OnScheduled public void parseColumns(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() < 0) { final DistributedMapCacheClient client = context.getProperty(DISTRIBUTED_CACHE_SERVICE).asControllerService(DistributedMapCacheClient.class); final ScanResult scanResult = getState(client); if (scanResult != null) { storeState(scanResult, context.getStateManager()); } clearState(client); } final String columnsValue = context.getProperty(COLUMNS).getValue(); final String[] columns = (columnsValue == null || columnsValue.isEmpty() ? new String[0] : columnsValue.split(",")); this.columns.clear(); for (final String column : columns) { if (column.contains(":")) { final String[] parts = column.split(":"); final byte[] cf = parts[0].getBytes(Charset.forName("UTF-8")); final byte[] cq = parts[1].getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, cq)); } else { final byte[] cf = column.getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, null)); } } } }
GetHBase extends AbstractProcessor { @OnScheduled public void parseColumns(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() < 0) { final DistributedMapCacheClient client = context.getProperty(DISTRIBUTED_CACHE_SERVICE).asControllerService(DistributedMapCacheClient.class); final ScanResult scanResult = getState(client); if (scanResult != null) { storeState(scanResult, context.getStateManager()); } clearState(client); } final String columnsValue = context.getProperty(COLUMNS).getValue(); final String[] columns = (columnsValue == null || columnsValue.isEmpty() ? new String[0] : columnsValue.split(",")); this.columns.clear(); for (final String column : columns) { if (column.contains(":")) { final String[] parts = column.split(":"); final byte[] cf = parts[0].getBytes(Charset.forName("UTF-8")); final byte[] cq = parts[1].getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, cq)); } else { final byte[] cf = column.getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, null)); } } }  }
GetHBase extends AbstractProcessor { @OnScheduled public void parseColumns(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() < 0) { final DistributedMapCacheClient client = context.getProperty(DISTRIBUTED_CACHE_SERVICE).asControllerService(DistributedMapCacheClient.class); final ScanResult scanResult = getState(client); if (scanResult != null) { storeState(scanResult, context.getStateManager()); } clearState(client); } final String columnsValue = context.getProperty(COLUMNS).getValue(); final String[] columns = (columnsValue == null || columnsValue.isEmpty() ? new String[0] : columnsValue.split(",")); this.columns.clear(); for (final String column : columns) { if (column.contains(":")) { final String[] parts = column.split(":"); final byte[] cf = parts[0].getBytes(Charset.forName("UTF-8")); final byte[] cq = parts[1].getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, cq)); } else { final byte[] cf = column.getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, null)); } } }  @Override Set<Relationship> getRelationships(); @Override void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue); @OnScheduled void parseColumns(final ProcessContext context); @OnPrimaryNodeStateChange void onPrimaryNodeChange(final PrimaryNodeState newState); @OnRemoved void onRemoved(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
GetHBase extends AbstractProcessor { @OnScheduled public void parseColumns(final ProcessContext context) throws IOException { final StateMap stateMap = context.getStateManager().getState(Scope.CLUSTER); if (stateMap.getVersion() < 0) { final DistributedMapCacheClient client = context.getProperty(DISTRIBUTED_CACHE_SERVICE).asControllerService(DistributedMapCacheClient.class); final ScanResult scanResult = getState(client); if (scanResult != null) { storeState(scanResult, context.getStateManager()); } clearState(client); } final String columnsValue = context.getProperty(COLUMNS).getValue(); final String[] columns = (columnsValue == null || columnsValue.isEmpty() ? new String[0] : columnsValue.split(",")); this.columns.clear(); for (final String column : columns) { if (column.contains(":")) { final String[] parts = column.split(":"); final byte[] cf = parts[0].getBytes(Charset.forName("UTF-8")); final byte[] cq = parts[1].getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, cq)); } else { final byte[] cf = column.getBytes(Charset.forName("UTF-8")); this.columns.add(new Column(cf, null)); } } }  @Override Set<Relationship> getRelationships(); @Override void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue); @OnScheduled void parseColumns(final ProcessContext context); @OnPrimaryNodeStateChange void onPrimaryNodeChange(final PrimaryNodeState newState); @OnRemoved void onRemoved(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session);  }