001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.replication;
019
020import java.util.ArrayList;
021import java.util.Collections;
022import java.util.List;
023import org.apache.hadoop.hbase.Cell;
024import org.apache.hadoop.hbase.HBaseInterfaceAudience;
025import org.apache.hadoop.hbase.wal.WAL.Entry;
026import org.apache.yetus.audience.InterfaceAudience;
027
028/**
029 * A {@link WALEntryFilter} which contains multiple filters and applies them in chain order
030 */
031@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.REPLICATION)
032public class ChainWALEntryFilter implements WALEntryFilter {
033
034  private final WALEntryFilter[] filters;
035  private WALCellFilter[] cellFilters;
036
037  public ChainWALEntryFilter(WALEntryFilter... filters) {
038    this.filters = filters;
039    initCellFilters();
040  }
041
042  public ChainWALEntryFilter(List<WALEntryFilter> filters) {
043    ArrayList<WALEntryFilter> rawFilters = new ArrayList<>(filters.size());
044    // flatten the chains
045    for (WALEntryFilter filter : filters) {
046      if (filter instanceof ChainWALEntryFilter) {
047        Collections.addAll(rawFilters, ((ChainWALEntryFilter) filter).filters);
048      } else {
049        rawFilters.add(filter);
050      }
051    }
052    this.filters = rawFilters.toArray(new WALEntryFilter[rawFilters.size()]);
053    initCellFilters();
054  }
055
056  public void initCellFilters() {
057    ArrayList<WALCellFilter> cellFilters = new ArrayList<>(filters.length);
058    for (WALEntryFilter filter : filters) {
059      if (filter instanceof WALCellFilter) {
060        cellFilters.add((WALCellFilter) filter);
061      }
062    }
063    this.cellFilters = cellFilters.toArray(new WALCellFilter[cellFilters.size()]);
064  }
065
066  @Override
067  public Entry filter(Entry entry) {
068    for (WALEntryFilter filter : filters) {
069      if (entry == null) {
070        return null;
071      }
072      entry = filter.filter(entry);
073    }
074    filterCells(entry);
075    return entry;
076  }
077
078  private void filterCells(Entry entry) {
079    if (entry == null || cellFilters.length == 0) {
080      return;
081    }
082    ArrayList<Cell> cells = entry.getEdit().getCells();
083    int size = cells.size();
084    for (int i = size - 1; i >= 0; i--) {
085      Cell cell = cells.get(i);
086      for (WALCellFilter filter : cellFilters) {
087        cell = filter.filterCell(entry, cell);
088        if (cell != null) {
089          cells.set(i, cell);
090        } else {
091          cells.remove(i);
092          break;
093        }
094      }
095    }
096    if (cells.size() < size / 2) {
097      cells.trimToSize();
098    }
099  }
100}