phoenix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jamestay...@apache.org
Subject [16/40] phoenix git commit: PHOENIX-4242 Fix Indexer post-compact hook logging of NPE and TableNotFound
Date Wed, 15 Nov 2017 18:34:44 GMT
PHOENIX-4242 Fix Indexer post-compact hook logging of NPE and TableNotFound


Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/632cbba8
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/632cbba8
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/632cbba8

Branch: refs/heads/4.x-HBase-1.2
Commit: 632cbba8914ed541c89ba4930b1b549709a237b9
Parents: 3b536f4
Author: Vincent Poon <vincentpoon@apache.org>
Authored: Thu Oct 19 14:28:27 2017 -0700
Committer: James Taylor <jtaylor@salesforce.com>
Committed: Wed Nov 15 10:02:13 2017 -0800

----------------------------------------------------------------------
 .../UngroupedAggregateRegionObserverIT.java     | 171 +++++++++++++++++++
 .../UngroupedAggregateRegionObserver.java       | 103 ++++++-----
 .../org/apache/phoenix/hbase/index/Indexer.java |  52 ------
 .../apache/phoenix/schema/MetaDataClient.java   |   3 +
 4 files changed, 239 insertions(+), 90 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/632cbba8/phoenix-core/src/it/java/org/apache/phoenix/end2end/UngroupedAggregateRegionObserverIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UngroupedAggregateRegionObserverIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UngroupedAggregateRegionObserverIT.java
new file mode 100644
index 0000000..3efd40e
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UngroupedAggregateRegionObserverIT.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.end2end;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.never;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.log4j.Appender;
+import org.apache.log4j.Level;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class UngroupedAggregateRegionObserverIT extends ParallelStatsDisabledIT {
+
+    private String dataTableName;
+    private String indexTableName;
+    private String schemaName;
+    private String dataTableFullName;
+    private static String indexTableFullName;
+
+    @Mock
+    private Appender mockAppender;
+
+    @Captor
+    private ArgumentCaptor<LoggingEvent> captorLoggingEvent;
+    private UngroupedAggregateRegionObserver ungroupedObserver;
+
+    @Before
+    public void setup() {
+        ungroupedObserver = new UngroupedAggregateRegionObserver();
+        ungroupedObserver.setCompactionConfig(PropertiesUtil.cloneConfig(config));
+    }
+
+    /**
+     * Tests the that post compact hook doesn't log any NPE for a System table
+     */
+    @Test
+    public void testPostCompactSystemSequence() throws Exception {
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            startCapturingIndexLog();
+            // run the post-compact hook
+            ungroupedObserver.clearTsOnDisabledIndexes("SYSTEM.SEQUENCE");
+            stopCapturingIndexLog();
+            // uneventful - nothing should be logged
+            Mockito.verify(mockAppender, never())
+                    .doAppend((LoggingEvent) captorLoggingEvent.capture());
+        }
+    }
+
+    /**
+     * Tests that calling the post compact hook on the data table permanently disables an
index that
+     * is being rebuilt (i.e. already disabled or inactive)
+     */
+    @Test
+    public void testPostCompactDataTableDuringRebuild() throws Exception {
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            generateUniqueTableNames();
+            testRebuildPostCompact(conn, dataTableFullName);
+        }
+    }
+
+    /**
+     * Tests that calling the post compact hook on the index table permanently disables an
index
+     * that is being rebuilt (i.e. already disabled or inactive)
+     */
+    @Test
+    public void testPostCompactIndexTableDuringRebuild() throws Exception {
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            generateUniqueTableNames();
+            testRebuildPostCompact(conn, indexTableFullName);
+        }
+    }
+
+    private void testRebuildPostCompact(Connection conn, String tableToCompact)
+            throws SQLException {
+        conn.createStatement().execute(
+            String.format(PartialScannerResultsDisabledIT.TEST_TABLE_DDL, dataTableFullName));
+        conn.createStatement().execute(String.format(PartialScannerResultsDisabledIT.INDEX_1_DDL,
+            indexTableName, dataTableFullName));
+        // disable the index, simulating an index write failure
+        PhoenixConnection pConn = conn.unwrap(PhoenixConnection.class);
+        IndexUtil.updateIndexState(pConn, indexTableFullName, PIndexState.DISABLE,
+            EnvironmentEdgeManager.currentTimeMillis());
+
+        // run the post-compact hook on the data table
+        startCapturingIndexLog();
+        ungroupedObserver.clearTsOnDisabledIndexes(tableToCompact);
+        stopCapturingIndexLog();
+        // an event should've been logged
+        Mockito.verify(mockAppender).doAppend((LoggingEvent) captorLoggingEvent.capture());
+        LoggingEvent loggingEvent = (LoggingEvent) captorLoggingEvent.getValue();
+        assertThat(loggingEvent.getLevel(), is(Level.INFO));
+        // index should be permanently disabled (disabletime of 0)
+        assertTrue(TestUtil.checkIndexState(pConn, indexTableFullName, PIndexState.DISABLE,
0L));
+    }
+
+    /**
+     * Tests that a non-Phoenix table (created purely through HBase) doesn't log a warning
in
+     * postCompact
+     */
+    @Test
+    public void testPostCompactTableNotFound() throws Exception {
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            HBaseTestingUtility utility = getUtility();
+            String nonPhoenixTable = "NOT_A_PHOENIX_TABLE";
+            utility.getHBaseAdmin().createTable(utility.createTableDescriptor(nonPhoenixTable));
+            startCapturingIndexLog();
+            ungroupedObserver.clearTsOnDisabledIndexes(nonPhoenixTable);
+            stopCapturingIndexLog();
+            // a debug level event should've been logged
+            Mockito.verify(mockAppender).doAppend((LoggingEvent) captorLoggingEvent.capture());
+            LoggingEvent loggingEvent = (LoggingEvent) captorLoggingEvent.getValue();
+            assertThat(loggingEvent.getLevel(), is(Level.DEBUG));
+        }
+    }
+
+    private void stopCapturingIndexLog() {
+        LogManager.getLogger(UngroupedAggregateRegionObserver.class).removeAppender(mockAppender);
+    }
+
+    private void startCapturingIndexLog() {
+        LogManager.getLogger(UngroupedAggregateRegionObserver.class).addAppender(mockAppender);
+    }
+
+    private void generateUniqueTableNames() {
+        schemaName = generateUniqueName();
+        dataTableName = generateUniqueName() + "_DATA";
+        dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName);
+        indexTableName = generateUniqueName() + "_IDX";
+        indexTableFullName = SchemaUtil.getTableName(schemaName, indexTableName);
+    }
+}

http://git-wip-us.apache.org/repos/asf/phoenix/blob/632cbba8/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index c3024a7..af50420 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -97,6 +97,7 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
 import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
 import org.apache.phoenix.index.IndexMaintainer;
 import org.apache.phoenix.index.PhoenixIndexCodec;
+import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
 import org.apache.phoenix.join.HashJoinInfo;
 import org.apache.phoenix.query.QueryConstants;
@@ -108,8 +109,10 @@ import org.apache.phoenix.schema.PIndexState;
 import org.apache.phoenix.schema.PRow;
 import org.apache.phoenix.schema.PTable;
 import org.apache.phoenix.schema.PTableImpl;
+import org.apache.phoenix.schema.PTableType;
 import org.apache.phoenix.schema.RowKeySchema;
 import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.schema.TableNotFoundException;
 import org.apache.phoenix.schema.TableRef;
 import org.apache.phoenix.schema.ValueSchema.Field;
 import org.apache.phoenix.schema.stats.StatisticsCollectionRunTracker;
@@ -133,7 +136,10 @@ import org.apache.phoenix.util.ExpressionUtil;
 import org.apache.phoenix.util.IndexUtil;
 import org.apache.phoenix.util.KeyValueUtil;
 import org.apache.phoenix.util.LogUtil;
+import org.apache.phoenix.util.MetaDataUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
 import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.QueryUtil;
 import org.apache.phoenix.util.ScanUtil;
 import org.apache.phoenix.util.SchemaUtil;
 import org.apache.phoenix.util.ServerUtil;
@@ -142,7 +148,10 @@ import org.apache.phoenix.util.TimeKeeper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Predicate;
 import com.google.common.base.Throwables;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import com.google.common.primitives.Ints;
@@ -926,7 +935,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
     }
 
     @Override
-    public void postCompact(final ObserverContext<RegionCoprocessorEnvironment> e,
final Store store,
+    public void postCompact(final ObserverContext<RegionCoprocessorEnvironment> c,
final Store store,
             final StoreFile resultFile, CompactionRequest request) throws IOException {
         // If we're compacting all files, then delete markers are removed
         // and we must permanently disable an index that needs to be
@@ -940,49 +949,67 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
             User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
                 @Override
                 public Void run() throws Exception {
-                    MutationCode mutationCode = null;
-                    long disableIndexTimestamp = 0;
-
-                    try (CoprocessorHConnection coprocessorHConnection =
-                            new CoprocessorHConnection(compactionConfig,
-                                    (HRegionServer) e.getEnvironment()
-                                            .getRegionServerServices());
-                            HTableInterface htable =
-                                    coprocessorHConnection
-                                            .getTable(SchemaUtil.getPhysicalTableName(
-                                                PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES,
-                                                compactionConfig))) {
-                        String tableName = e.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
-                        // FIXME: if this is an index on a view, we won't find a row for
it in SYSTEM.CATALOG
-                        // Instead, we need to disable all indexes on the view.
-                        byte[] tableKey = SchemaUtil.getTableKeyFromFullName(tableName);
-                        Get get = new Get(tableKey);
-                        get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES);
-                        Result result = htable.get(get);
-                        if (!result.isEmpty()) {
-                            Cell cell = result.listCells().get(0);
-                            if (cell.getValueLength() > 0) {
-                                disableIndexTimestamp = PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(),
cell.getValueOffset(), SortOrder.getDefault());
-                                if (disableIndexTimestamp != 0) {
-                                    logger.info("Major compaction running while index on
table is disabled.  Clearing index disable timestamp: " + tableName);
-                                    mutationCode = IndexUtil.updateIndexState(tableKey, 0L,
htable, PIndexState.DISABLE).getMutationCode();
-                                }
-                            }
-                        }
-                    } catch (Throwable t) { // log, but swallow exception as we don't want
to impact compaction
-                        logger.warn("Potential failure to permanently disable index during
compaction " +  e.getEnvironment().getRegionInfo().getTable().getNameAsString(), t);
-                    } finally {
-                        if (disableIndexTimestamp != 0 && mutationCode != MutationCode.TABLE_ALREADY_EXISTS
&& mutationCode != MutationCode.TABLE_NOT_FOUND) {
-                            logger.warn("Attempt to permanently disable index " + e.getEnvironment().getRegionInfo().getTable().getNameAsString()
+ 
-                                    " during compaction" + (mutationCode == null ? "" : "
failed with code = " + mutationCode));
-                        }
-                    }
+                    String fullTableName = c.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
+                    clearTsOnDisabledIndexes(fullTableName);
                     return null;
                 }
             });
         }
     }
 
+    @VisibleForTesting
+    public void clearTsOnDisabledIndexes(final String fullTableName) {
+        try (PhoenixConnection conn =
+                QueryUtil.getConnectionOnServer(compactionConfig).unwrap(PhoenixConnection.class))
{
+            String baseTable = fullTableName;
+            PTable table = PhoenixRuntime.getTableNoCache(conn, baseTable);
+            List<PTable> indexes;
+            // if it's an index table, we just need to check if it's disabled
+            if (PTableType.INDEX.equals(table.getType())) {
+                indexes = Lists.newArrayList(table.getIndexes());
+                indexes.add(table);
+            } else {
+                // for a data table, check all its indexes
+                indexes = table.getIndexes();
+            }
+            // FIXME need handle views and indexes on views as well
+            // if any index is disabled, we won't have all the data for a rebuild after compaction
+            for (PTable index : indexes) {
+                if (index.getIndexDisableTimestamp() != 0) {
+                    try {
+                        logger.info(
+                            "Major compaction running while index on table is disabled. 
Clearing index disable timestamp: "
+                                    + index);
+                        IndexUtil.updateIndexState(conn, index.getName().getString(),
+                            PIndexState.DISABLE, Long.valueOf(0L));
+                    } catch (SQLException e) {
+                        logger.warn(
+                            "Unable to permanently disable index " + index.getName().getString(),
+                            e);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            if (e instanceof TableNotFoundException) {
+                logger.debug("Ignoring HBase table that is not a Phoenix table: " + fullTableName);
+                // non-Phoenix HBase tables won't be found, do nothing
+                return;
+            }
+            // If we can't reach the stats table, don't interrupt the normal
+            // compaction operation, just log a warning.
+            if (logger.isWarnEnabled()) {
+                logger.warn("Unable to permanently disable indexes being partially rebuild
for "
+                        + fullTableName,
+                    e);
+            }
+        }
+    }
+
+    @VisibleForTesting
+    public void setCompactionConfig(Configuration compactionConfig) {
+        this.compactionConfig = compactionConfig;
+    }
+
     private static PTable deserializeTable(byte[] b) {
         try {
             PTableProtos.PTable ptableProto = PTableProtos.PTable.parseFrom(b);

http://git-wip-us.apache.org/repos/asf/phoenix/blob/632cbba8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index 4ac4ab5..9686789 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -25,7 +25,6 @@ import static org.apache.phoenix.hbase.index.write.IndexWriterUtils.INDEX_WRITER
 
 import java.io.IOException;
 import java.security.PrivilegedExceptionAction;
-import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -62,8 +61,6 @@ import org.apache.hadoop.hbase.regionserver.OperationStatus;
 import org.apache.hadoop.hbase.regionserver.Region;
 import org.apache.hadoop.hbase.regionserver.ScanType;
 import org.apache.hadoop.hbase.regionserver.Store;
-import org.apache.hadoop.hbase.regionserver.StoreFile;
-import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
 import org.apache.hadoop.hbase.security.User;
@@ -89,18 +86,12 @@ import org.apache.phoenix.hbase.index.write.IndexWriter;
 import org.apache.phoenix.hbase.index.write.RecoveryIndexWriter;
 import org.apache.phoenix.hbase.index.write.recovery.PerRegionIndexWriteCache;
 import org.apache.phoenix.hbase.index.write.recovery.StoreFailuresInCachePolicy;
-import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.schema.PIndexState;
-import org.apache.phoenix.schema.PTable;
 import org.apache.phoenix.trace.TracingUtils;
 import org.apache.phoenix.trace.util.NullSpan;
 import org.apache.phoenix.util.EnvironmentEdgeManager;
-import org.apache.phoenix.util.IndexUtil;
-import org.apache.phoenix.util.PhoenixRuntime;
 import org.apache.phoenix.util.PropertiesUtil;
-import org.apache.phoenix.util.QueryUtil;
 import org.apache.phoenix.util.ServerUtil;
 
 import com.google.common.collect.Lists;
@@ -847,48 +838,5 @@ public class Indexer extends BaseRegionObserver {
     properties.put(Indexer.INDEX_BUILDER_CONF_KEY, builder.getName());
     desc.addCoprocessor(Indexer.class.getName(), null, priority, properties);
   }
-  
-  @Override
-  public void postCompact(final ObserverContext<RegionCoprocessorEnvironment> c, final
Store store,
-          final StoreFile resultFile, CompactionRequest request) throws IOException {
-      // If we're compacting all files, then delete markers are removed
-      // and we must permanently disable an index that needs to be
-      // partially rebuild because we're potentially losing the information
-      // we need to successfully rebuilt it.
-      if (request.isAllFiles() || request.isMajor()) {
-          // Compaction and split upcalls run with the effective user context of the requesting
user.
-          // This will lead to failure of cross cluster RPC if the effective user is not
-          // the login user. Switch to the login user context to ensure we have the expected
-          // security context.
-          User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
-              @Override
-              public Void run() throws Exception {
-                  String fullTableName = c.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
-                  try {
-                      PhoenixConnection conn =  QueryUtil.getConnectionOnServer(compactionConfig).unwrap(PhoenixConnection.class);
-                      PTable table = PhoenixRuntime.getTableNoCache(conn, fullTableName);
-                      // FIXME: we may need to recurse into children of this table too
-                      for (PTable index : table.getIndexes()) {
-                          if (index.getIndexDisableTimestamp() != 0) {
-                              try {
-                                  LOG.info("Major compaction running while index on table
is disabled.  Clearing index disable timestamp: " + fullTableName);
-                                  IndexUtil.updateIndexState(conn, index.getName().getString(),
PIndexState.DISABLE, Long.valueOf(0L));
-                              } catch (SQLException e) {
-                                  LOG.warn("Unable to permanently disable index " + index.getName().getString(),
e);
-                              }
-                          }
-                      }
-                  } catch (Exception e) {
-                      // If we can't reach the stats table, don't interrupt the normal
-                      // compaction operation, just log a warning.
-                      if (LOG.isWarnEnabled()) {
-                          LOG.warn("Unable to permanently disable indexes being partially
rebuild for " + fullTableName, e);
-                      }
-                  }
-                  return null;
-              }
-          });
-      }
-  }
 }
 

http://git-wip-us.apache.org/repos/asf/phoenix/blob/632cbba8/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 0f6bab2..0ce4246 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -602,6 +602,9 @@ public class MetaDataClient {
             }
 
             if (SYSTEM_CATALOG_SCHEMA.equals(schemaName)) {
+                if (result.getMutationCode() == MutationCode.TABLE_ALREADY_EXISTS &&
result.getTable() == null) {
+                    result.setTable(table);
+                }
                 return result;
             }
             MutationCode code = result.getMutationCode();


Mime
View raw message