Skip to content

HBASE-23309: Adding the flexibility to ChainWalEntryFilter to filter the whole entry if all cells get filtered #837

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
xcangCRM merged 9 commits into from
Dec 6, 2019
Merged

HBASE-23309: Adding the flexibility to ChainWalEntryFilter to filter the whole entry if all cells get filtered #837

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
7e627ff
Fixing the ChainWalEntryFilter to filter the whole entry if all cells…
Nov 17, 2019
9faebc9
Adding tests for WALCellFilter to cover all the scenarios
Nov 17, 2019
ce10dc1
Merge branch 'master' of github.com:apache/hbase into HBASE-23309
Nov 17, 2019
1b16c3a
Adding the filter empty entry into peer configuration
Nov 18, 2019
ffb78d3
Removing the set filter empty entry from the replication peer config …
Nov 22, 2019
02fdfec
nit fix
Nov 22, 2019
cd3f767
Adding an InterfaceAudience
Nov 22, 2019
fec27df
Addressing comments and fixing style checking
Dec 2, 2019
5b28db9
fixing check style
Dec 3, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1295,6 +1295,9 @@ public enum OperationStatusCode {
/** Configuration key for SplitLog manager timeout */
public static final String HBASE_SPLITLOG_MANAGER_TIMEOUT = "hbase.splitlog.manager.timeout";

/** To allow the empty entries to get filtered which have no cells or all cells got filtered though WALCellFilter */
public static final String HBASE_REPLICATION_WAL_FILTER_EMPTY_ENTRY = "hbase.replication.wal.filteremptyentry";

@apurtell apurtell Nov 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't belong here. You are already changing BaseReplicationEndpoint, put it there?

@xcangCRM xcangCRM Dec 6, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@sandeepvinayak Do you plan addressing this comment here? I just saw this before I tried to merge this. Thanks.


/**
* Configuration keys for Bucket cache
*/
Expand Down
9 changes: 5 additions & 4 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/replication/BaseReplicationEndpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,15 +41,16 @@ public abstract class BaseReplicationEndpoint extends AbstractService
public static final String REPLICATION_WALENTRYFILTER_CONFIG_KEY
= "hbase.replication.source.custom.walentryfilters";
protected Context ctx;
private ReplicationPeer replicationPeer;

@wchevreuil wchevreuil Nov 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why this needs to be a global variable? Seems to be used only on init method.


@Override
public void init(Context context) throws IOException {
this.ctx = context;

if (this.ctx != null){
ReplicationPeer peer = this.ctx.getReplicationPeer();
if (peer != null){
peer.registerPeerConfigListener(this);
replicationPeer = this.ctx.getReplicationPeer();
if (replicationPeer != null){
replicationPeer.registerPeerConfigListener(this);
} else {
LOG.warn("Not tracking replication peer config changes for Peer Id " + this.ctx.getPeerId() +
" because there's no such peer");
Expand Down Expand Up @@ -91,7 +92,7 @@ public WALEntryFilter getWALEntryfilter() {
}
}
}
return filters.isEmpty() ? null : new ChainWALEntryFilter(filters);
return filters.isEmpty() ? null : new ChainWALEntryFilter(filters, this.replicationPeer);

@apurtell apurtell Nov 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there some other way to do this that does not require adding ReplicationPeer to the method signature?

Why not create a WALEntryFilter that drops empty cells? Then you can add it to the chain if you want and no further changes are needed.

}

/** Returns a WALEntryFilter for checking the scope. Subclasses can
Expand Down
25 changes: 24 additions & 1 deletion hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ChainWALEntryFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.yetus.audience.InterfaceAudience;

import static org.apache.hadoop.hbase.HConstants.HBASE_REPLICATION_WAL_FILTER_EMPTY_ENTRY;

/**
* A {@link WALEntryFilter} which contains multiple filters and applies them
* in chain order
Expand All @@ -36,13 +38,25 @@ public class ChainWALEntryFilter implements WALEntryFilter {

private final WALEntryFilter[] filters;
private WALCellFilter[] cellFilters;
private ReplicationPeerConfig peerConfig;

// To allow the empty entries to get filtered, we want to this optional flag to decide
// if we want to filter the entries which have no cells or all cells got filtered though WALCellFilter
private String filterEmptyEntry;

public ChainWALEntryFilter(WALEntryFilter...filters) {
this.filters = filters;
initCellFilters();
}

public ChainWALEntryFilter(List<WALEntryFilter> filters) {
public ChainWALEntryFilter(List<WALEntryFilter> filters, ReplicationPeer replicationPeer) {
if (replicationPeer != null) {
peerConfig = replicationPeer.getPeerConfig();
if (peerConfig != null) {
filterEmptyEntry = peerConfig.getConfiguration().get(HBASE_REPLICATION_WAL_FILTER_EMPTY_ENTRY);
}
}

ArrayList<WALEntryFilter> rawFilters = new ArrayList<>(filters.size());
// flatten the chains
for (WALEntryFilter filter : filters) {
Expand All @@ -68,13 +82,17 @@ public void initCellFilters() {

@Override
public Entry filter(Entry entry) {

Comment thread
sandeepvinayak marked this conversation as resolved.
for (WALEntryFilter filter : filters) {
if (entry == null) {
return null;
}
entry = filter.filter(entry);
}
filterCells(entry);
if (shouldFilterEmptyEntry() && entry != null && entry.getEdit().isEmpty()) {

@apurtell apurtell Nov 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Put this into your own WALEntryFilter. Or, fine if you want to add a WALEntryFilter impl to HBase code that does this. But don't do this here

@sandeepvinayak sandeepvinayak Nov 19, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@apurtell This is the flexility we want to provide to custom replication endpoint so that every custom replication endpoint doesn't need to re-implement everything what ChainWALEntryFilter already does.

Here is the scenario we want to cover, let's take an example:

CustomWALFilter implements WALEntryFilter, WALCellFilter {
    @override 
     public void filter(Entry) {
    }
    @override 
    public void filterCell(Entry, Cell){
    }
}

Custom Replication endpoint set the filters by:

ChainWALEntryFilter(filters); \\ new CustomWALFilter() is part of filters

if filter in the above CustomWALFilter returns Entry but filterCell filters all the cells, the ChainWALEntryFilter will not return null in current implementation. Since most of the CustomWALFilter uses ChainWALEntryFilter, isn't it better to provide this flexibility in ChainWALFilter itself? Sure, it's possible to all this logic in CustomWALFilter, but shouldn't we provide re-usability through ChainWALFilter and provides a flexibility through a config? Let me know your thoughts.

@wchevreuil wchevreuil Nov 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this additional logic would fit better in a built-in filter (one of @apurtell suggestions above). That would not need the extra config property nor changes in BaseRepplicationEndpoint

return null;
}
return entry;
}

Expand All @@ -94,4 +112,9 @@ private Cell filterCell(Entry entry, Cell cell) {
}
return cell;
}

public boolean shouldFilterEmptyEntry() {
return (filterEmptyEntry != null
&& filterEmptyEntry.equalsIgnoreCase("true"));
}
}
2 changes: 1 addition & 1 deletion ...ver/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,7 +301,7 @@ private void initializeWALEntryFilter(UUID peerClusterId) {
filters.add(filterFromEndpoint);
}
filters.add(new ClusterMarkingEntryFilter(clusterId, peerClusterId, replicationEndpoint));
this.walEntryFilter = new ChainWALEntryFilter(filters);
this.walEntryFilter = new ChainWALEntryFilter(filters, this.replicationPeer);
}

private void tryStartNewShipper(String walGroupId, PriorityBlockingQueue<Path> queue) {
Expand Down
49 changes: 49 additions & 0 deletions ...ver/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationWALEntryFilters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
import org.junit.experimental.categories.Category;

import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.mockito.Mockito;

@Category({ ReplicationTests.class, SmallTests.class })
public class TestReplicationWALEntryFilters {
Expand Down Expand Up @@ -155,6 +156,54 @@ public Entry filter(Entry entry) {
}
};

public static class FilterSomeCellsWALCellFilter implements WALEntryFilter, WALCellFilter {
@Override
public Entry filter(Entry entry) {
return entry;
}

@Override
public Cell filterCell(Entry entry, Cell cell) {
if (Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()).equals("a")) {
return null;
} else {
return cell;
}
}
}

public static class FilterAllCellsWALCellFilter implements WALEntryFilter, WALCellFilter {
@Override
public Entry filter(Entry entry) {
return entry;
}

@Override
public Cell filterCell(Entry entry, Cell cell) {
return null;
}
}

@Test
public void testChainWALEntryWithCellFilter() {
Entry userEntry = createEntry(null, a, b, c);
ChainWALEntryFilter filterSomeCells = new ChainWALEntryFilter(new FilterSomeCellsWALCellFilter());
// since WALCellFilter filter cells with rowkey 'a'
assertEquals(createEntry(null, b,c), filterSomeCells.filter(userEntry));

Entry userEntry2 = createEntry(null, b, c, d);
// since there is no cell to get filtered, nothing should get filtered
assertEquals(userEntry2, filterSomeCells.filter(userEntry2));

ChainWALEntryFilter filterAllCells = new ChainWALEntryFilter(new FilterAllCellsWALCellFilter());
ChainWALEntryFilter spyFilterAllCells = Mockito.spy(filterAllCells);
assertEquals(createEntry(null), spyFilterAllCells.filter(userEntry));
// let's set the filter empty entry flag to true now for the above case
Mockito.doReturn(true).when(spyFilterAllCells).shouldFilterEmptyEntry();
// since WALCellFilter filter all cells, whole entry should be filtered
assertEquals(null, spyFilterAllCells.filter(userEntry));
}

@Test
public void testChainWALEntryFilter() {
Entry userEntry = createEntry(null, a, b, c);
Expand Down