[
https://issues.apache.org/jira/browse/PHOENIX-5980?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17777370#comment-17777370
]
ASF GitHub Bot commented on PHOENIX-5980:
-----------------------------------------
gjacoby126 commented on code in PR #1717:
URL: https://github.com/apache/phoenix/pull/1717#discussion_r1365844868
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithAllDeleteIT.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithAllDeleteIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithAllDeleteIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
+ createTables();
+ populateTables();
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(3);
+ serverProps.put("hbase.coprocessor.abortonerror", "false");
+ serverProps.put(Indexer.CHECK_VERSION_CONF_KEY, "false");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS,
String.valueOf(true));
+ clientProps.put(QueryServices.TRANSACTIONS_ENABLED, "true");
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
+
+ }
+
+ @Parameterized.Parameters(name =
"MutationBatchFailedStateMetricWithAllDeleteIT_transactionProvider={0}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { "OMID" }, { null } });
+ }
+
+ /**
+ * We will have 5 rows in table and we will try deleting 4 out of 5 rows
In this test intention
+ * is to fail the delete and see how metric value is updated when that
happens for transactional
+ * table and non transactional table
+ * @throws SQLException
+ */
+ @Test
+ public void testFailedDelete() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = null;
+ try {
+ conn = DriverManager.getConnection(getUrl(), props);
+
+ Statement stmt = conn.createStatement();
+ // adding coprocessor which basically overrides prebatchmutate
whiich is called for all
+ // the mutations
+ // it is called before applying mutation to region
+ TestUtil.addCoprocessor(conn, deleteTableName,
+ ImmutableIndexIT.DeleteFailingRegionObserver.class);
+ // trying to delete 4 rows with this single delete statement
+ String dml = String.format("DELETE FROM %s where val1 > 1",
deleteTableName);
+ stmt.execute(dml);
+ conn.commit();
+
+ throw new AssertionError("Commit should not have succeeded");
+ } catch (SQLException e) {
+ Map<String, Map<MetricType, Long>> mutationMetrics =
+ conn.unwrap(PhoenixConnection.class).getMutationMetrics();
+ int mfs =
+
mutationMetrics.get(deleteTableName).get(MUTATION_BATCH_FAILED_SIZE).intValue();
+ int dbfs =
+
mutationMetrics.get(deleteTableName).get(DELETE_BATCH_FAILED_SIZE).intValue();
+ int upfs =
+
mutationMetrics.get(deleteTableName).get(UPSERT_BATCH_FAILED_SIZE).intValue();
+ long gfs =
+ GLOBAL_MUTATION_BATCH_FAILED_COUNT.getMetric().getValue();
+
+ Assert.assertEquals(4, dbfs);
+ Assert.assertEquals(4, mfs);
+ Assert.assertEquals(0, upfs);
+ //for the second parameter the global mutation batch failed count
is double the original parameter because
Review Comment:
Rather than assuming an execution order, might be cleaner just to keep track
of which parameter (1 or 2) we're on and multiply it by 4.
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithAllDeleteIT.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithAllDeleteIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithAllDeleteIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
Review Comment:
Please remove system.out.println calls
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithAllUpsertIT.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithAllUpsertIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
Review Comment:
There's a lot of common boilerplate between the three new ITs. How difficult
would it be to consolidate to one, or have a common superclass?
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithDeleteAndUpsertIT.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithDeleteAndUpsertIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
Review Comment:
What do the Delete and Upsert ITs added above show that this IT (which does
both delete and upsert) doesn't?
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithAllUpsertIT.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithAllUpsertIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithAllUpsertIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
+ createTables();
+ populateTables();
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(3);
+ serverProps.put("hbase.coprocessor.abortonerror", "false");
+ serverProps.put(Indexer.CHECK_VERSION_CONF_KEY, "false");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS,
String.valueOf(true));
+ clientProps.put(QueryServices.TRANSACTIONS_ENABLED, "true");
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
+
+ }
+
+ @Parameterized.Parameters(name =
"MutationBatchFailedStateMetricWithAllUpsertIT_transactionProvider={0}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { "OMID" }, { null } });
+ }
+
+ /**
+ * We will try upserting 1 rows In this test intention is to fail the
upsert and
+ * see how metric value is updated when that happens for transactional
+ * table and non-transactional table
+ * @throws SQLException
+ */
+ @Test
+ public void testFailedUpsert() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = null;
+ try {
+ conn = DriverManager.getConnection(getUrl(), props);
+
+ Statement stmt = conn.createStatement();
+ // adding coprocessor which basically overrides prebatchmutate
whiich is called for all
+ // the mutations
+ // it is called before applying mutation to region
+ TestUtil.addCoprocessor(conn, deleteTableName,
+ ImmutableIndexIT.DeleteFailingRegionObserver.class);
+
+ String upsertSQL = String.format(upsertStatement, deleteTableName);
+ PreparedStatement preparedStatement =
conn.prepareStatement(upsertSQL);
+ preparedStatement.setString(1, "ROW_6");
+ preparedStatement.setInt(2, 6);
+ preparedStatement.setInt(3, 12);
+ preparedStatement.execute();
+
+ conn.commit();
+
+ throw new AssertionError("Commit should not have succeeded");
Review Comment:
Assert.fail()
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithAllDeleteIT.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithAllDeleteIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithAllDeleteIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
+ createTables();
+ populateTables();
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(3);
+ serverProps.put("hbase.coprocessor.abortonerror", "false");
+ serverProps.put(Indexer.CHECK_VERSION_CONF_KEY, "false");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS,
String.valueOf(true));
+ clientProps.put(QueryServices.TRANSACTIONS_ENABLED, "true");
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
+
+ }
+
+ @Parameterized.Parameters(name =
"MutationBatchFailedStateMetricWithAllDeleteIT_transactionProvider={0}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { "OMID" }, { null } });
+ }
+
+ /**
+ * We will have 5 rows in table and we will try deleting 4 out of 5 rows
In this test intention
+ * is to fail the delete and see how metric value is updated when that
happens for transactional
+ * table and non transactional table
+ * @throws SQLException
+ */
+ @Test
+ public void testFailedDelete() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = null;
+ try {
+ conn = DriverManager.getConnection(getUrl(), props);
+
+ Statement stmt = conn.createStatement();
+ // adding coprocessor which basically overrides prebatchmutate
whiich is called for all
+ // the mutations
+ // it is called before applying mutation to region
+ TestUtil.addCoprocessor(conn, deleteTableName,
+ ImmutableIndexIT.DeleteFailingRegionObserver.class);
+ // trying to delete 4 rows with this single delete statement
+ String dml = String.format("DELETE FROM %s where val1 > 1",
deleteTableName);
+ stmt.execute(dml);
+ conn.commit();
+
+ throw new AssertionError("Commit should not have succeeded");
Review Comment:
nit: You can use Assert.fail() here.
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithDeleteAndUpsertIT.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithDeleteAndUpsertIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithDeleteAndUpsertIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
+ createTables();
+ populateTables();
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(3);
+ serverProps.put("hbase.coprocessor.abortonerror", "false");
+ serverProps.put(Indexer.CHECK_VERSION_CONF_KEY, "false");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS,
String.valueOf(true));
+ clientProps.put(QueryServices.TRANSACTIONS_ENABLED, "true");
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
+
+ }
+
+ @Parameterized.Parameters(name =
"MutationBatchFailedStateMetricWithDeleteAndUpsertIT_transactionProvider={0}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { "OMID" }, { null } });
+ }
+
+
+ @Test
+ public void testFailedUpsertAndDelete() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = null;
+
+ try {
+ conn = DriverManager.getConnection(getUrl(), props);
+
+ Statement stmt = conn.createStatement();
+ // adding coprocessor which basically overrides prebatchmutate
whiich is called for all
+ // the mutations
+ // it is called before applying mutation to region
+ TestUtil.addCoprocessor(conn, deleteTableName,
+ ImmutableIndexIT.DeleteFailingRegionObserver.class);
+ // trying to delete 4 rows with this single delete statement
+ String dml = String.format("DELETE FROM %s where val1 > 1",
deleteTableName);
+ stmt.execute(dml);
+
+ String upsertSQL = String.format(upsertStatement, deleteTableName);
+ PreparedStatement preparedStatement =
conn.prepareStatement(upsertSQL);
+ preparedStatement.setString(1, "ROW_6");
+ preparedStatement.setInt(2, 6);
+ preparedStatement.setInt(3, 12);
+ preparedStatement.execute();
+
+ conn.commit();
+
+ throw new AssertionError("Commit should not have succeeded");
+ } catch (SQLException e) {
+ Map<String, Map<MetricType, Long>> mutationMetrics =
+ conn.unwrap(PhoenixConnection.class).getMutationMetrics();
+ int mfs =
+
mutationMetrics.get(deleteTableName).get(MUTATION_BATCH_FAILED_SIZE).intValue();
+ int dbfs =
+
mutationMetrics.get(deleteTableName).get(DELETE_BATCH_FAILED_SIZE).intValue();
+ int upfs =
+
mutationMetrics.get(deleteTableName).get(UPSERT_BATCH_FAILED_SIZE).intValue();
+ long gfs =
+ GLOBAL_MUTATION_BATCH_FAILED_COUNT.getMetric().getValue();
+
+
+ Assert.assertEquals(4, dbfs);
+ Assert.assertEquals(5, mfs);
+ Assert.assertEquals(1, upfs);
+ //for the second parameter the global mutation batch failed count
is double the original parameter because
+ //global metrics do not reset to 0 for each parameter
+ if(this.transactionProvider == null){
Review Comment:
Likewise we're assuming an order here when we really care about whether this
is the first or second parameter.
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithDeleteAndUpsertIT.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithDeleteAndUpsertIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithDeleteAndUpsertIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
+ createTables();
+ populateTables();
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(3);
+ serverProps.put("hbase.coprocessor.abortonerror", "false");
+ serverProps.put(Indexer.CHECK_VERSION_CONF_KEY, "false");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
+ clientProps.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS,
String.valueOf(true));
+ clientProps.put(QueryServices.TRANSACTIONS_ENABLED, "true");
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
+
+ }
+
+ @Parameterized.Parameters(name =
"MutationBatchFailedStateMetricWithDeleteAndUpsertIT_transactionProvider={0}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { "OMID" }, { null } });
+ }
+
+
+ @Test
+ public void testFailedUpsertAndDelete() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = null;
+
+ try {
+ conn = DriverManager.getConnection(getUrl(), props);
+
+ Statement stmt = conn.createStatement();
+ // adding coprocessor which basically overrides prebatchmutate
whiich is called for all
+ // the mutations
+ // it is called before applying mutation to region
+ TestUtil.addCoprocessor(conn, deleteTableName,
+ ImmutableIndexIT.DeleteFailingRegionObserver.class);
+ // trying to delete 4 rows with this single delete statement
+ String dml = String.format("DELETE FROM %s where val1 > 1",
deleteTableName);
+ stmt.execute(dml);
+
+ String upsertSQL = String.format(upsertStatement, deleteTableName);
+ PreparedStatement preparedStatement =
conn.prepareStatement(upsertSQL);
+ preparedStatement.setString(1, "ROW_6");
+ preparedStatement.setInt(2, 6);
+ preparedStatement.setInt(3, 12);
+ preparedStatement.execute();
+
+ conn.commit();
+
+ throw new AssertionError("Commit should not have succeeded");
Review Comment:
Assert.fail()
##########
phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java:
##########
@@ -1584,12 +1588,26 @@ public static MutationMetricQueue.MutationMetric
updateMutationBatchFailureMetri
numDeleteMutationsInBatch++;
}
}
+
+ long totalFailedMutation = numUpsertMutationsInBatch +
numDeleteMutationsInBatch;
+ //TODO: this case is something that should not happen
Review Comment:
Comment seems valid but it doesn't seem like a TODO.
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/MutationBatchFailedStateMetricWithAllUpsertIT.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static
org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
+
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.phoenix.end2end.index.ImmutableIndexIT;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+
+@Category(NeedsOwnMiniClusterTest.class)
+@RunWith(Parameterized.class)
+public class MutationBatchFailedStateMetricWithAllUpsertIT extends
ParallelStatsDisabledIT {
+ String create_table =
+ "CREATE TABLE IF NOT EXISTS %s(ID VARCHAR NOT NULL PRIMARY KEY,
VAL1 INTEGER, VAL2 INTEGER)";
+ String indexName = generateUniqueName();
+ String create_index = "CREATE INDEX " + indexName + " ON %s(VAL1 DESC)
INCLUDE (VAL2)";
+ String upsertStatement = "UPSERT INTO %s VALUES(?, ?, ?)";
+ String deleteTableName = generateUniqueName();
+ private final boolean transactional;
+ private String transactionProvider;
+
+ public MutationBatchFailedStateMetricWithAllUpsertIT(String
transactionProvider) {
+ this.transactional = transactionProvider != null;
+ if (this.transactional) {
+ create_table =
+ create_table + (this.transactional
+ ? (" TRANSACTIONAL=true,TRANSACTION_PROVIDER='" +
transactionProvider
+ + "'")
+ : "");
+ }
+ this.transactionProvider = transactionProvider;
+ System.out.println("transaction provider " + this.transactionProvider);
Review Comment:
please remove System.out.println calls
> MUTATION_BATCH_FAILED_SIZE metric is incorrectly updated for failing delete
> mutations
> -------------------------------------------------------------------------------------
>
> Key: PHOENIX-5980
> URL: https://issues.apache.org/jira/browse/PHOENIX-5980
> Project: Phoenix
> Issue Type: Bug
> Affects Versions: 4.15.0
> Reporter: Chinmay Kulkarni
> Assignee: Divneet Kaur
> Priority: Major
> Labels: metrics, phoenix-hardening, quality-improvement
>
> In the conn.commit() path, we get the number of mutations that failed to be
> committed in the catch block of MutationState.sendMutations() (see
> [here|https://github.com/apache/phoenix/blob/dcc88af8acc2ba8df10d2e9d498ab3646fdf0a78/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java#L1195-L1198]).
>
> In case of delete mutations, the uncommittedStatementIndexes.length always
> resolves to 1 and we always update the metric value by 1 in this case, even
> though the actual mutation list corresponds to multiple DELETE mutations
> which failed. In case of upserts, using unCommittedStatementIndexes.length is
> fine since each upsert query corresponds to 1 Put. We should fix the logic
> for deletes/mixed delete + upsert mutation batch failures.
> This wrong value is propagated to global client metrics as well as
> MutationMetricQueue metrics.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
