IndexerEditFilter implements WALEditFilter { @Override public void apply(WALEdit walEdit) { List<KeyValue> keyValues = walEdit.getKeyValues(); for (int i = keyValues.size() - 1; i >= 0; i--) { if (!isValidKeyValue(keyValues.get(i))) { keyValues.remove(i); } } } IndexerEditFilter(String subscriptionName); @Override void apply(WALEdit walEdit); String getSubscriptionName(); static final String NO_INDEX_FLAG; }
@Test public void testApply_NoPayload() { WALEdit walEdit = new WALEdit(); walEdit.add(new KeyValue(Bytes.toBytes("row"), RecordCf.DATA.bytes, Bytes.toBytes("not payload"), Bytes.toBytes("value"))); editFilter.apply(walEdit); assertEquals(0, walEdit.size()); }
@Test public void testApply_Payload_NotApplicableIndex() { RecordEvent recordEvent = new RecordEvent(); IndexRecordFilterData filterData = new IndexRecordFilterData(); filterData.setSubscriptionInclusions(ImmutableSet.of("SomeOtherIndexName")); recordEvent.setIndexRecordFilterData(filterData); WALEdit walEdit = new WALEdit(); walEdit.add(new KeyValue(Bytes.toBytes("row"), RecordCf.DATA.bytes, RecordColumn.PAYLOAD.bytes, recordEvent.toJsonBytes())); editFilter.apply(walEdit); assertEquals(0, walEdit.size()); }
@Test public void testApply_Payload_ApplicableIndex() { RecordEvent recordEvent = new RecordEvent(); IndexRecordFilterData filterData = new IndexRecordFilterData(); filterData.setSubscriptionInclusions(ImmutableSet.of(INDEX_NAME)); recordEvent.setIndexRecordFilterData(filterData); WALEdit walEdit = new WALEdit(); walEdit.add(new KeyValue(Bytes.toBytes("row"), RecordCf.DATA.bytes, RecordColumn.PAYLOAD.bytes, recordEvent.toJsonBytes())); editFilter.apply(walEdit); assertEquals(1, walEdit.size()); }
@Test public void testApply_Payload_ApplicableIndexButNoIndexFlagIsSet() { RecordEvent recordEvent = new RecordEvent(); IndexRecordFilterData filterData = new IndexRecordFilterData(); filterData.setSubscriptionInclusions(ImmutableSet.of(INDEX_NAME)); recordEvent.setIndexRecordFilterData(filterData); recordEvent.getAttributes().put(IndexerEditFilter.NO_INDEX_FLAG, "false"); WALEdit walEdit = new WALEdit(); walEdit.add(new KeyValue(Bytes.toBytes("row"), RecordCf.DATA.bytes, RecordColumn.PAYLOAD.bytes, recordEvent.toJsonBytes())); editFilter.apply(walEdit); assertEquals(0, walEdit.size()); }
@Test public void testApply_NonJsonPayload() { WALEdit walEdit = new WALEdit(); walEdit.add(new KeyValue(Bytes.toBytes("row"), RecordCf.DATA.bytes, RecordColumn.PAYLOAD.bytes, Bytes.toBytes("non-json payload"))); editFilter.apply(walEdit); assertEquals(0, walEdit.size()); }