[GitHub] incubator-tephra issue #74: TEPHRA-266 Identify log messages when multiple i...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/74 This has been merged to master branch ---
[GitHub] incubator-tephra pull request #74: TEPHRA-266 Identify log messages when mul...
Github user poornachandra closed the pull request at: https://github.com/apache/incubator-tephra/pull/74 ---
[GitHub] incubator-tephra pull request #72: TEPHRA-270 TEPHRA-271 Transaction state c...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/72#discussion_r185069321
--- Diff:
tephra-core/src/main/java/org/apache/tephra/coprocessor/TransactionStateCache.java
---
@@ -78,6 +78,7 @@ protected void startUp() throws Exception {
protected void shutDown() throws Exception {
if (refreshService != null) {
refreshService.interrupt();
+ refreshService.join(1000);
--- End diff --
`Thread.join(long millis)` does not throw an exception on timeout being
reached. It only throws `IllegalArgumentException` when timeout is negative or
`InterruptedException`.
Are you talking about handling the `InterruptedException`?
---
[GitHub] incubator-tephra pull request #74: TEPHRA-266 Identify log messages when mul...
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/74 TEPHRA-266 Identify log messages when multiple instances of Tephra run on a single HBase cluster JIRA - https://issues.apache.org/jira/browse/TEPHRA-266 You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/tx-state-cache-log Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/74.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #74 commit df16b5be2146c4ae92d23d2c62a2cf3b32a0ced5 Author: poorna <poorna@...> Date: 2018-04-26T01:58:52Z TEPHRA-266 Identify log messages when multiple instances of Tephra run on a single HBase cluster ---
[GitHub] incubator-tephra pull request #73: TEPHRA-267 Save the service reference in ...
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/73 TEPHRA-267 Save the service reference in ReferenceCountedSupplier only if startup is successful JIRA - https://issues.apache.org/jira/browse/TEPHRA-267 You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/ref-count-supplier Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/73.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #73 commit 01f172157eb959f5a2e52d8d7c38f948298f30c1 Author: poorna <poorna@...> Date: 2018-04-26T00:15:48Z TEPHRA-267 Save the service reference in ReferenceCountedSupplier only if startup is successful ---
[GitHub] incubator-tephra pull request #72: TEPHRA-270 TEPHRA-271 Transaction state c...
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/72 TEPHRA-270 TEPHRA-271 Transaction state cache bug fixes JIRA: https://issues.apache.org/jira/browse/TEPHRA-270 https://issues.apache.org/jira/browse/TEPHRA-271 This PR contains a couple of bug fixes for the Transaction state cache to fix race conditions and deadlocks. You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/tx-state-cache Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/72.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #72 commit 73feed9a6a93d032c2e19e4e7d77fa4eb2065fcf Author: poorna <poorna@...> Date: 2018-04-25T23:51:56Z TEPHRA-270 Do not refresh the transaction state during co-processor startup to prevent deadlocks commit d42b282ec03c8e0d07bbb2a714961f2dd9b7641d Author: poorna <poorna@...> Date: 2018-04-25T23:58:22Z TEPHRA-271 Wait for the refresh thread to stop during shutdown to prevent race conditions ---
[GitHub] incubator-tephra issue #71: (TEPHRA-275) Exclude failing test from Travis bu...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/71 @anew It would be good to commit this against JIRA TEPHRA-285, looks like you intended to use that but ended up using TEPHRA-275 ---
[GitHub] incubator-tephra issue #67: TEPHRA-272 Add HBase 2.0 compatibility module
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/67 @ankitsinghal I spent some time on the tephra-hbase-compat-1.4 compat module tests failure. It looks like the 1.4 compat module tests were never run earlier. When you changed the travis.xml to build hbase-compat-1.4 module, the issue was exposed. I have filed JIRA https://issues.apache.org/jira/browse/TEPHRA-285 for this issue. Since the failure was already present, I will go ahead and merge this PR. Thanks for the contribution! ---
[GitHub] incubator-tephra issue #63: TEPHRA-265 Fix NOTICE_BINARY for Guice and Guice...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/63 @johnament We are planning for the next release of Tephra soon. Can you take a look at the changes, so that this PR can be part of the next release? ---
[GitHub] incubator-tephra issue #67: TEPHRA-272 Add HBase 2.0 compatibility module
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/67 @joshelser There is nothing stopping us from moving Tephra to Java 8. Just that we'll have to build consensus on the Tephra dev list, and it will add to the delay of this PR. If making the POM changes to work with both Java 7 and Java 8 is complicated, then @ankitsinghal can you send out an email to the dev list for the Java 8 consensus? ---
[GitHub] incubator-tephra issue #67: TEPHRA-272 Add HBase 2.0 compatibility module
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/67 @ankitsinghal Since HBase-2.0 needs Java 8, let's try changing the compiler for only HBase-2.0 compat module to use Java 8. We can still keep the other modules as Java 7. I haven't tried it out myself. But I think it should work. Right now, we run Travis tests using both Java 7 and Java 8 for all modules. We can restrict HBase-2.0 compat module tests to run only for Java 8. ---
[GitHub] incubator-tephra issue #67: TEPHRA-272 Add HBase 2.0 compatibility module
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/67 @ankitsinghal Just one more question (https://github.com/apache/incubator-tephra/pull/67#discussion_r169503415). Also the Java 7 build is failing for HBase 2.0 compat module. Looks like HBase 2.0 needs Java 8. Can you update the pom for the 2.0 compat module to use Java 8? Note that the master branch has some changes on how the Travis tests are run (TEPHRA-282). You'll have to rebase, and then add an entry to travis.yml for the 2.0 compat module. ---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r169503415
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/coprocessor/FilteredInternalScanner.java
---
@@ -0,0 +1,80 @@
+/*
+ * 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.tephra.hbase.coprocessor;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.regionserver.ScannerContext;
+import
org.apache.tephra.hbase.coprocessor.TransactionProcessor.IncludeInProgressFilter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Wrapper of InternalScanner to apply Transaction visibility filter for
flush and compact
+ */
+public class FilteredInternalScanner implements InternalScanner {
+
+ private InternalScanner delegate;
+ private Filter filter;
+
+ public FilteredInternalScanner(InternalScanner internalScanner,
IncludeInProgressFilter filter) {
+this.delegate = internalScanner;
+this.filter = filter;
+ }
+
+ @Override
+ public void close() throws IOException {
+this.delegate.close();
+ }
+
+ @Override
+ public boolean next(List result, ScannerContext scannerContext)
throws IOException {
+List outResult = new ArrayList();
+while (true) {
+ boolean next = delegate.next(outResult, scannerContext);
+ for (Cell cell : outResult) {
+ReturnCode code = filter.filterKeyValue(cell);
+switch (code) {
+// included, so we are done
+case INCLUDE:
+case INCLUDE_AND_NEXT_COL:
--- End diff --
The only downside I see with this approach is that we'll iterate through
all the cells for a column, even though we could have skipped some cells when
the filter returns `INCLUDE_AND_NEXT_COL`.
For example if there are 100 versions of a value in a column, and the
filter figures out at the 10th version that the rest of the versions have
reached the TTL, then in the earlier code we would not iterate through the
rest of the 90 cells (or at least that's what I think was happening in the
underlying scanner). Is there anyway we can achieve the same result with the
new APIs?
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r169195010
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/coprocessor/FilteredInternalScanner.java
---
@@ -0,0 +1,80 @@
+/*
+ * 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.tephra.hbase.coprocessor;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.regionserver.ScannerContext;
+import
org.apache.tephra.hbase.coprocessor.TransactionProcessor.IncludeInProgressFilter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Wrapper of InternalScanner to apply Transaction visibility filter for
flush and compact
+ */
+public class FilteredInternalScanner implements InternalScanner {
+
+ private InternalScanner delegate;
+ private Filter filter;
+
+ public FilteredInternalScanner(InternalScanner internalScanner,
IncludeInProgressFilter filter) {
+this.delegate = internalScanner;
+this.filter = filter;
+ }
+
+ @Override
+ public void close() throws IOException {
+this.delegate.close();
+ }
+
+ @Override
+ public boolean next(List result, ScannerContext scannerContext)
throws IOException {
+List outResult = new ArrayList();
+while (true) {
+ boolean next = delegate.next(outResult, scannerContext);
+ for (Cell cell : outResult) {
+ReturnCode code = filter.filterKeyValue(cell);
+switch (code) {
+// included, so we are done
+case INCLUDE:
+case INCLUDE_AND_NEXT_COL:
--- End diff --
ð Just confirmed that transaction filter does this
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r16679
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/coprocessor/FilteredInternalScanner.java
---
@@ -0,0 +1,80 @@
+/*
+ * 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.tephra.hbase.coprocessor;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.regionserver.ScannerContext;
+import
org.apache.tephra.hbase.coprocessor.TransactionProcessor.IncludeInProgressFilter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Wrapper of InternalScanner to apply Transaction visibility filter for
flush and compact
+ */
+public class FilteredInternalScanner implements InternalScanner {
+
+ private InternalScanner delegate;
+ private Filter filter;
+
+ public FilteredInternalScanner(InternalScanner internalScanner,
IncludeInProgressFilter filter) {
+this.delegate = internalScanner;
+this.filter = filter;
+ }
+
+ @Override
+ public void close() throws IOException {
+this.delegate.close();
+ }
+
+ @Override
+ public boolean next(List result, ScannerContext scannerContext)
throws IOException {
+List outResult = new ArrayList();
--- End diff --
It would be good to make `outResult` as a private variable, so that it is
not created for every `next` call. We can then just clear it for each call.
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r16822
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/coprocessor/FilteredInternalScanner.java
---
@@ -0,0 +1,80 @@
+/*
+ * 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.tephra.hbase.coprocessor;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.regionserver.ScannerContext;
+import
org.apache.tephra.hbase.coprocessor.TransactionProcessor.IncludeInProgressFilter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Wrapper of InternalScanner to apply Transaction visibility filter for
flush and compact
+ */
+public class FilteredInternalScanner implements InternalScanner {
+
+ private InternalScanner delegate;
+ private Filter filter;
--- End diff --
Both the fields can be marked as final since they don't change.
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160400229
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/TransactionAwareHTable.java
---
@@ -0,0 +1,729 @@
+/*
+ * 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.tephra.hbase;
+
+import com.google.protobuf.Descriptors.MethodDescriptor;
+import com.google.protobuf.Message;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.CompareOperator;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+import org.apache.tephra.AbstractTransactionAwareTable;
+import org.apache.tephra.Transaction;
+import org.apache.tephra.TransactionAware;
+import org.apache.tephra.TxConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A Transaction Aware HTable implementation for HBase 2.0. Operations are
committed as usual, but
+ * upon a failed or aborted transaction, they are rolled back to the state
before the transaction
+ * was started.
+ */
+public class TransactionAwareHTable extends AbstractTransactionAwareTable
+implements Table, TransactionAware {
+
+private static final Logger LOG =
LoggerFactory.getLogger(TransactionAwareHTable.class);
+private final Table hTable;
+
+/**
+ * Create a transactional aware instance of the passed HTable
+ * @param hTable underlying HBase table to use
+ */
+public TransactionAwareHTable(Table hTable) {
+this(hTable, false);
+}
+
+/**
+ * Create a transactional aware instance of the passed HTable
+ * @param hTable underlying HBase table to use
+ * @param conflictLevel level of conflict detection to perform
(defaults to {@code COLUMN})
+ */
+public TransactionAwareHTable(Table hTable,
TxConstants.ConflictDetection conflictLevel) {
+this(hTable, conflictLevel, false);
+}
+
+/**
+ * Create a transactional aware instance of the passed HTable, with
the option of allowing
+ * non-transactional operations.
+ * @param hTable underlying HBase table to use
+ * @param allowNonTransactional if true, additional operations
(checkAndPut, increment,
+ *checkAndDelete) will be available, though
non-transactional
+ */
+public TransactionAwareHTable(T
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/67#discussion_r160405134 --- Diff: tephra-hbase-compat-2.0/src/test/java/org/apache/tephra/hbase/coprocessor/TransactionProcessorTest.java --- @@ -0,0 +1,677 @@ +/* + * 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.tephra.hbase.coprocessor; + +import com.google.common.collect.ImmutableSortedMap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; + +import it.unimi.dsi.fastutil.longs.LongArrayList; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.ChoreService; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionInfo; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.MockRegionServerServices; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.TableDescriptorBuilder; +import org.apache.hadoop.hbase.io.util.MemorySizeUtil; +import org.apache.hadoop.hbase.regionserver.ChunkCreator; +import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker; +import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResult; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResultImpl; +import org.apache.hadoop.hbase.regionserver.HRegionFileSystem; +import org.apache.hadoop.hbase.regionserver.MemStoreLAB; +import org.apache.hadoop.hbase.regionserver.MemStoreLABImpl; +import org.apache.hadoop.hbase.regionserver.RegionScanner; +import org.apache.hadoop.hbase.regionserver.ScanType; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.wal.WAL; +import org.apache.hadoop.hbase.wal.WALFactory; +import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.tephra.Transaction; +import org.apache.tephra.TransactionManager; +import org.apache.tephra.TxConstants; +import org.apache.tephra.coprocessor.TransactionStateCache; +import org.apache.tephra.coprocessor.TransactionStateCacheSupplier; +import org.apache.tephra.manager.InvalidTxList; +import org.apache.tephra.metrics.TxMetricsCollector; +import org.apache.tephra.persist.HDFSTransactionStateStorage; +import org.apache.tephra.persist.TransactionSnapshot; +import org.apache.tephra.persist.TransactionVisibilityState; +import org.apache.tephra.snapshot.DefaultSnapshotCodec; +import org.apache.tephra.snapshot.SnapshotCodecProvider; +import org.apache.tephra.util.TxUtils; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/67#discussion_r160406273 --- Diff: tephra-hbase-compat-2.0/src/test/java/org/apache/tephra/hbase/coprocessor/TransactionProcessorTest.java --- @@ -0,0 +1,677 @@ +/* + * 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.tephra.hbase.coprocessor; + +import com.google.common.collect.ImmutableSortedMap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; + +import it.unimi.dsi.fastutil.longs.LongArrayList; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.ChoreService; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionInfo; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.MockRegionServerServices; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.TableDescriptorBuilder; +import org.apache.hadoop.hbase.io.util.MemorySizeUtil; +import org.apache.hadoop.hbase.regionserver.ChunkCreator; +import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker; +import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResult; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResultImpl; +import org.apache.hadoop.hbase.regionserver.HRegionFileSystem; +import org.apache.hadoop.hbase.regionserver.MemStoreLAB; +import org.apache.hadoop.hbase.regionserver.MemStoreLABImpl; +import org.apache.hadoop.hbase.regionserver.RegionScanner; +import org.apache.hadoop.hbase.regionserver.ScanType; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.wal.WAL; +import org.apache.hadoop.hbase.wal.WALFactory; +import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.tephra.Transaction; +import org.apache.tephra.TransactionManager; +import org.apache.tephra.TxConstants; +import org.apache.tephra.coprocessor.TransactionStateCache; +import org.apache.tephra.coprocessor.TransactionStateCacheSupplier; +import org.apache.tephra.manager.InvalidTxList; +import org.apache.tephra.metrics.TxMetricsCollector; +import org.apache.tephra.persist.HDFSTransactionStateStorage; +import org.apache.tephra.persist.TransactionSnapshot; +import org.apache.tephra.persist.TransactionVisibilityState; +import org.apache.tephra.snapshot.DefaultSnapshotCodec; +import org.apache.tephra.snapshot.SnapshotCodecProvider; +import org.apache.tephra.util.TxUtils; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160397436
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -0,0 +1,182 @@
+/*
+ * 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.tephra.hbase;
+
+import com.google.common.base.Throwables;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.tephra.TransactionContext;
+import org.apache.tephra.TransactionFailureException;
+import org.apache.tephra.distributed.TransactionServiceClient;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A Transactional SecondaryIndexTable.
+ */
+public class SecondaryIndexTable implements Closeable {
+ private byte[] secondaryIndex;
+ private TransactionAwareHTable transactionAwareHTable;
+ private TransactionAwareHTable secondaryIndexTable;
+ private TransactionContext transactionContext;
+ private final TableName secondaryIndexTableName;
+ private Connection connection;
+ private static final byte[] secondaryIndexFamily =
Bytes.toBytes("secondaryIndexFamily");
+ private static final byte[] secondaryIndexQualifier = Bytes.toBytes('r');
+ private static final byte[] DELIMITER = new byte[] {0};
+
+ public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, Table table,
+ byte[] secondaryIndex) throws IOException {
+secondaryIndexTableName =
TableName.valueOf(table.getName().getNameAsString() + ".idx");
+this.connection =
ConnectionFactory.createConnection(table.getConfiguration());
+Table secondaryIndexHTable = null;
+try (Admin hBaseAdmin = this.connection.getAdmin()) {
+ if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
+
hBaseAdmin.createTable(TableDescriptorBuilder.newBuilder(secondaryIndexTableName).build());
+ }
+ secondaryIndexHTable =
this.connection.getTable(secondaryIndexTableName);
+} catch (Exception e) {
+ Throwables.propagate(e);
--- End diff --
It would be good to close the `connection` in case of exception.
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/67#discussion_r160407023 --- Diff: tephra-hbase-compat-2.0/src/test/java/org/apache/tephra/hbase/coprocessor/TransactionProcessorTest.java --- @@ -0,0 +1,677 @@ +/* + * 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.tephra.hbase.coprocessor; + +import com.google.common.collect.ImmutableSortedMap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; + +import it.unimi.dsi.fastutil.longs.LongArrayList; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.ChoreService; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionInfo; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.MockRegionServerServices; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.TableDescriptorBuilder; +import org.apache.hadoop.hbase.io.util.MemorySizeUtil; +import org.apache.hadoop.hbase.regionserver.ChunkCreator; +import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker; +import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResult; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResultImpl; +import org.apache.hadoop.hbase.regionserver.HRegionFileSystem; +import org.apache.hadoop.hbase.regionserver.MemStoreLAB; +import org.apache.hadoop.hbase.regionserver.MemStoreLABImpl; +import org.apache.hadoop.hbase.regionserver.RegionScanner; +import org.apache.hadoop.hbase.regionserver.ScanType; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.wal.WAL; +import org.apache.hadoop.hbase.wal.WALFactory; +import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.tephra.Transaction; +import org.apache.tephra.TransactionManager; +import org.apache.tephra.TxConstants; +import org.apache.tephra.coprocessor.TransactionStateCache; +import org.apache.tephra.coprocessor.TransactionStateCacheSupplier; +import org.apache.tephra.manager.InvalidTxList; +import org.apache.tephra.metrics.TxMetricsCollector; +import org.apache.tephra.persist.HDFSTransactionStateStorage; +import org.apache.tephra.persist.TransactionSnapshot; +import org.apache.tephra.persist.TransactionVisibilityState; +import org.apache.tephra.snapshot.DefaultSnapshotCodec; +import org.apache.tephra.snapshot.SnapshotCodecProvider; +import org.apache.tephra.util.TxUtils; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160402855
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java
---
@@ -0,0 +1,583 @@
+/*
+ * 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.tephra.hbase.coprocessor;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionObserver;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterBase;
+import org.apache.hadoop.hbase.regionserver.HStore;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.ScanOptions;
+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.CompactionLifeCycleTracker;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.tephra.Transaction;
+import org.apache.tephra.TransactionCodec;
+import org.apache.tephra.TxConstants;
+import org.apache.tephra.coprocessor.CacheSupplier;
+import org.apache.tephra.coprocessor.TransactionStateCache;
+import org.apache.tephra.coprocessor.TransactionStateCacheSupplier;
+import org.apache.tephra.hbase.txprune.CompactionState;
+import org.apache.tephra.persist.TransactionVisibilityState;
+import org.apache.tephra.util.TxUtils;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.Nullable;
+
+/**
+ * {@code org.apache.hadoop.hbase.coprocessor.RegionObserver} coprocessor
that handles server-side processing
+ * for transactions:
+ *
+ * applies filtering to exclude data from invalid and in-progress
transactions
+ * overrides the scanner returned for flush and compaction to drop
data written by invalidated transactions,
+ * or expired due to TTL.
+ *
+ *
+ * In order to use this coprocessor for transactions, configure the
class on any table involved in transactions,
+ * or on all user tables by adding the following to hbase-site.xml:
+ * {@code
+ *
+ * hbase.coprocessor.region.classes
+ *
org.apache.tephra.hbase.coprocessor.TransactionProcessor
+ *
+ * }
+ *
+ *
+ * HBase {@code Get} and {@code Scan} operations should have the
current tr
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/67#discussion_r160406873 --- Diff: tephra-hbase-compat-2.0/src/test/java/org/apache/tephra/hbase/coprocessor/TransactionProcessorTest.java --- @@ -0,0 +1,677 @@ +/* + * 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.tephra.hbase.coprocessor; + +import com.google.common.collect.ImmutableSortedMap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; + +import it.unimi.dsi.fastutil.longs.LongArrayList; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.ChoreService; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionInfo; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.MockRegionServerServices; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.TableDescriptorBuilder; +import org.apache.hadoop.hbase.io.util.MemorySizeUtil; +import org.apache.hadoop.hbase.regionserver.ChunkCreator; +import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker; +import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResult; +import org.apache.hadoop.hbase.regionserver.HRegion.FlushResultImpl; +import org.apache.hadoop.hbase.regionserver.HRegionFileSystem; +import org.apache.hadoop.hbase.regionserver.MemStoreLAB; +import org.apache.hadoop.hbase.regionserver.MemStoreLABImpl; +import org.apache.hadoop.hbase.regionserver.RegionScanner; +import org.apache.hadoop.hbase.regionserver.ScanType; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.wal.WAL; +import org.apache.hadoop.hbase.wal.WALFactory; +import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.tephra.Transaction; +import org.apache.tephra.TransactionManager; +import org.apache.tephra.TxConstants; +import org.apache.tephra.coprocessor.TransactionStateCache; +import org.apache.tephra.coprocessor.TransactionStateCacheSupplier; +import org.apache.tephra.manager.InvalidTxList; +import org.apache.tephra.metrics.TxMetricsCollector; +import org.apache.tephra.persist.HDFSTransactionStateStorage; +import org.apache.tephra.persist.TransactionSnapshot; +import org.apache.tephra.persist.TransactionVisibilityState; +import org.apache.tephra.snapshot.DefaultSnapshotCodec; +import org.apache.tephra.snapshot.SnapshotCodecProvider; +import org.apache.tephra.util.TxUtils; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160398165
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -0,0 +1,182 @@
+/*
+ * 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.tephra.hbase;
+
+import com.google.common.base.Throwables;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.tephra.TransactionContext;
+import org.apache.tephra.TransactionFailureException;
+import org.apache.tephra.distributed.TransactionServiceClient;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A Transactional SecondaryIndexTable.
+ */
+public class SecondaryIndexTable implements Closeable {
+ private byte[] secondaryIndex;
+ private TransactionAwareHTable transactionAwareHTable;
+ private TransactionAwareHTable secondaryIndexTable;
+ private TransactionContext transactionContext;
+ private final TableName secondaryIndexTableName;
+ private Connection connection;
+ private static final byte[] secondaryIndexFamily =
Bytes.toBytes("secondaryIndexFamily");
+ private static final byte[] secondaryIndexQualifier = Bytes.toBytes('r');
+ private static final byte[] DELIMITER = new byte[] {0};
+
+ public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, Table table,
+ byte[] secondaryIndex) throws IOException {
+secondaryIndexTableName =
TableName.valueOf(table.getName().getNameAsString() + ".idx");
+this.connection =
ConnectionFactory.createConnection(table.getConfiguration());
+Table secondaryIndexHTable = null;
+try (Admin hBaseAdmin = this.connection.getAdmin()) {
+ if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
+
hBaseAdmin.createTable(TableDescriptorBuilder.newBuilder(secondaryIndexTableName).build());
+ }
+ secondaryIndexHTable =
this.connection.getTable(secondaryIndexTableName);
+} catch (Exception e) {
+ Throwables.propagate(e);
+}
+
+this.secondaryIndex = secondaryIndex;
+this.transactionAwareHTable = new TransactionAwareHTable(table);
+this.secondaryIndexTable = new
TransactionAwareHTable(secondaryIndexHTable);
+this.transactionContext = new
TransactionContext(transactionServiceClient, transactionAwareHTable,
+ secondaryIndexTable);
+ }
+
+ public Result get(Get get) throws IOException {
+return get(Collections.singletonList(get))[0];
+ }
+
+ public Result[] get(List gets) throws IOException {
+try {
+ transactionContext.start();
+ Result[] result = transactionAwareHTable.get(gets);
+ transactionContext.finish();
+ return result;
+} catch (Exception e) {
+ try {
+transactionContext.abort();
+ } catch (TransactionFailureException e1) {
+throw new IOException("Could not rollback transaction", e1);
+ }
+}
+return null
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160403830
--- Diff:
tephra-hbase-compat-2.0/src/test/java/org/apache/tephra/hbase/AbstractHBaseTableTest.java
---
@@ -0,0 +1,106 @@
+/*
+ * 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.tephra.hbase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Coprocessor;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.tephra.TxConstants;
+import org.apache.tephra.hbase.coprocessor.TransactionProcessor;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Base class for tests that need a HBase cluster
+ */
+@SuppressWarnings("WeakerAccess")
+public abstract class AbstractHBaseTableTest {
+ protected static HBaseTestingUtility testUtil;
+ protected static HBaseAdmin hBaseAdmin;
+ protected static Configuration conf;
+
+ @BeforeClass
+ public static void startMiniCluster() throws Exception {
+testUtil = conf == null ? new HBaseTestingUtility() : new
HBaseTestingUtility(conf);
+conf = testUtil.getConfiguration();
+
+// Tune down the connection thread pool size
+conf.setInt("hbase.hconnection.threads.core", 5);
+conf.setInt("hbase.hconnection.threads.max", 10);
+// Tunn down handler threads in regionserver
+conf.setInt("hbase.regionserver.handler.count", 10);
+
+// Set to random port
+conf.setInt("hbase.master.port", 0);
+conf.setInt("hbase.master.info.port", 0);
+conf.setInt("hbase.regionserver.port", 0);
+conf.setInt("hbase.regionserver.info.port", 0);
+
+testUtil.startMiniCluster();
+hBaseAdmin = testUtil.getHBaseAdmin();
+ }
+
+ @AfterClass
+ public static void shutdownMiniCluster() throws Exception {
+try {
+ if (hBaseAdmin != null) {
+hBaseAdmin.close();
+ }
+} finally {
+ testUtil.shutdownMiniCluster();
+}
+ }
+
+ protected static Table createTable(byte[] tableName, byte[][]
columnFamilies) throws Exception {
+return createTable(tableName, columnFamilies, false,
+
Collections.singletonList(TransactionProcessor.class.getName()));
+ }
+
+ protected static Table createTable(byte[] tableName, byte[][]
columnFamilies, boolean existingData,
+ List coprocessors) throws
Exception {
+HTableDescriptor desc = new
HTableDescriptor(TableName.valueOf(tableName));
+for (byte[] family : columnFamilies) {
+ HColumnDescriptor columnDesc = new HColumnDescriptor(family);
+ columnDesc.setMaxVersions(Integer.MAX_VALUE);
+ columnDesc.setValue(TxConstants.PROPERTY_TTL,
String.valueOf(10)); // in millis
+ desc.addFamily(columnDesc);
+}
+if (existingData) {
+ desc.setValue(TxConstants.READ_NON_TX_DATA, "true");
+}
+// Divide individually to prevent any overflow
+int priority = Coprocessor.PRIORITY_USER;
+// order in list is the same order that coprocessors will be invoked
+for (String coprocessor : coprocessors) {
+ desc.addCoprocessor(coprocessor, null, ++priority, null);
+}
+hBaseAdmin.createTable(desc);
+//testUtil.waitTableAvailable(tableName, 5000);
--- End diff --
Is the wait no longer required here?
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160400034
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/TransactionAwareHTable.java
---
@@ -0,0 +1,729 @@
+/*
+ * 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.tephra.hbase;
+
+import com.google.protobuf.Descriptors.MethodDescriptor;
+import com.google.protobuf.Message;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.CompareOperator;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+import org.apache.tephra.AbstractTransactionAwareTable;
+import org.apache.tephra.Transaction;
+import org.apache.tephra.TransactionAware;
+import org.apache.tephra.TxConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A Transaction Aware HTable implementation for HBase 2.0. Operations are
committed as usual, but
+ * upon a failed or aborted transaction, they are rolled back to the state
before the transaction
+ * was started.
+ */
+public class TransactionAwareHTable extends AbstractTransactionAwareTable
+implements Table, TransactionAware {
+
+private static final Logger LOG =
LoggerFactory.getLogger(TransactionAwareHTable.class);
+private final Table hTable;
+
+/**
+ * Create a transactional aware instance of the passed HTable
+ * @param hTable underlying HBase table to use
+ */
+public TransactionAwareHTable(Table hTable) {
+this(hTable, false);
+}
+
+/**
+ * Create a transactional aware instance of the passed HTable
+ * @param hTable underlying HBase table to use
+ * @param conflictLevel level of conflict detection to perform
(defaults to {@code COLUMN})
+ */
+public TransactionAwareHTable(Table hTable,
TxConstants.ConflictDetection conflictLevel) {
+this(hTable, conflictLevel, false);
+}
+
+/**
+ * Create a transactional aware instance of the passed HTable, with
the option of allowing
+ * non-transactional operations.
+ * @param hTable underlying HBase table to use
+ * @param allowNonTransactional if true, additional operations
(checkAndPut, increment,
+ *checkAndDelete) will be available, though
non-transactional
+ */
+public TransactionAwareHTable(T
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160401932
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/TransactionAwareHTable.java
---
@@ -0,0 +1,729 @@
+/*
+ * 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.tephra.hbase;
+
+import com.google.protobuf.Descriptors.MethodDescriptor;
+import com.google.protobuf.Message;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.CompareOperator;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+import org.apache.tephra.AbstractTransactionAwareTable;
+import org.apache.tephra.Transaction;
+import org.apache.tephra.TransactionAware;
+import org.apache.tephra.TxConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A Transaction Aware HTable implementation for HBase 2.0. Operations are
committed as usual, but
+ * upon a failed or aborted transaction, they are rolled back to the state
before the transaction
+ * was started.
+ */
+public class TransactionAwareHTable extends AbstractTransactionAwareTable
+implements Table, TransactionAware {
+
+private static final Logger LOG =
LoggerFactory.getLogger(TransactionAwareHTable.class);
+private final Table hTable;
+
+/**
+ * Create a transactional aware instance of the passed HTable
+ * @param hTable underlying HBase table to use
+ */
+public TransactionAwareHTable(Table hTable) {
+this(hTable, false);
+}
+
+/**
+ * Create a transactional aware instance of the passed HTable
+ * @param hTable underlying HBase table to use
+ * @param conflictLevel level of conflict detection to perform
(defaults to {@code COLUMN})
+ */
+public TransactionAwareHTable(Table hTable,
TxConstants.ConflictDetection conflictLevel) {
+this(hTable, conflictLevel, false);
+}
+
+/**
+ * Create a transactional aware instance of the passed HTable, with
the option of allowing
+ * non-transactional operations.
+ * @param hTable underlying HBase table to use
+ * @param allowNonTransactional if true, additional operations
(checkAndPut, increment,
+ *checkAndDelete) will be available, though
non-transactional
+ */
+public TransactionAwareHTable(T
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160404092
--- Diff:
tephra-hbase-compat-2.0/src/test/java/org/apache/tephra/hbase/TransactionAwareHTableTest.java
---
@@ -0,0 +1,1907 @@
+/*
+ * 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.tephra.hbase;
+
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionObserver;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.filter.LongComparator;
+import org.apache.hadoop.hbase.filter.ValueFilter;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.tephra.Transaction;
+import org.apache.tephra.TransactionConflictException;
+import org.apache.tephra.TransactionContext;
+import org.apache.tephra.TransactionManager;
+import org.apache.tephra.TransactionSystemClient;
+import org.apache.tephra.TxConstants;
+import org.apache.tephra.TxConstants.ConflictDetection;
+import org.apache.tephra.hbase.coprocessor.TransactionProcessor;
+import org.apache.tephra.inmemory.InMemoryTxSystemClient;
+import org.apache.tephra.metrics.TxMetricsCollector;
+import org.apache.tephra.persist.HDFSTransactionStateStorage;
+import org.apache.tephra.persist.TransactionStateStorage;
+import org.apache.tephra.snapshot.SnapshotCodecProvider;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+
+/**
+ * Tests for TransactionAwareHTables.
+ */
+public class TransactionAwareHTableTest extends AbstractHBaseTableTest {
+private static final Logger LOG =
LoggerFactory.getLogger(TransactionAwareHTableTest.class);
--- End diff --
This file also needs to use 2 space for indentation
---
[GitHub] incubator-tephra pull request #67: TEPHRA-272 Add HBase 2.0 compatibility mo...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/67#discussion_r160398641
--- Diff:
tephra-hbase-compat-2.0/src/main/java/org/apache/tephra/hbase/TransactionAwareHTable.java
---
@@ -0,0 +1,729 @@
+/*
+ * 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.tephra.hbase;
+
+import com.google.protobuf.Descriptors.MethodDescriptor;
+import com.google.protobuf.Message;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.CompareOperator;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.OperationWithAttributes;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+import org.apache.tephra.AbstractTransactionAwareTable;
+import org.apache.tephra.Transaction;
+import org.apache.tephra.TransactionAware;
+import org.apache.tephra.TxConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A Transaction Aware HTable implementation for HBase 2.0. Operations are
committed as usual, but
+ * upon a failed or aborted transaction, they are rolled back to the state
before the transaction
+ * was started.
+ */
+public class TransactionAwareHTable extends AbstractTransactionAwareTable
+implements Table, TransactionAware {
+
+private static final Logger LOG =
LoggerFactory.getLogger(TransactionAwareHTable.class);
--- End diff --
The convention in Tephra is to use 2 spaces for indentation, this class
uses 4 spaces.
---
[GitHub] incubator-tephra issue #67: TEPHRA-272 Add HBase 2.0 compatibility module
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/67 @ankitsinghal I'm taking a look at the PR. Meanwhile, the Travis build has failed. Looks like there is some issue finding HBase 2.0 dependencies. Can you take a look at it? Also, I'm on vacation this month. So there may be a delay in my replies. ---
[GitHub] incubator-tephra issue #63: TEPHRA-265 Fix NOTICE_BINARY for Guice and Guice...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/63 @gokulavasan Do you have time to take this PR to completion? ---
[GitHub] incubator-tephra issue #45: [TEPHRA-236] Changes to replace deprecated class...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/45 Thanks for the contribution @bijugs! ---
[GitHub] incubator-tephra issue #62: TEPHRA-245 Improve prune debug tool
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/62 @anew I have ported over the changes to other comapt modules, please take a look ---
[GitHub] incubator-tephra pull request #62: TEPHRA-245 Improve prune debug tool
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/62#discussion_r140183601
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/txprune/InvalidListPruningDebugTool.java
---
@@ -204,16 +204,19 @@ public int compare(RegionPruneInfo o1,
RegionPruneInfo o2) {
return Long.compare(o1.getPruneUpperBound(),
o2.getPruneUpperBound());
}
};
-Queue lowestPrunes =
+MinMaxPriorityQueue lowestPrunes =
MinMaxPriorityQueue.orderedBy(comparator).maximumSize(numRegions).create();
for (RegionPruneInfo pruneInfo : regionPruneInfos) {
lowestPrunes.add(new RegionPruneInfoPretty(pruneInfo));
}
-TreeSet regionSet = new TreeSet<>(comparator);
-regionSet.addAll(lowestPrunes);
-return regionSet;
+List regions = new ArrayList<>(numRegions);
+RegionPruneInfoPretty e;
+while ((e = lowestPrunes.pollFirst()) != null) {
--- End diff --
I noticed that `addAll()` can use a for loop to iterate over the elements
of the priority queue, and adds the element to the destination collection.
The javadoc for the `MinMaxPriorityQueue.iterator()` says -
```
Returns an iterator over the elements contained in this collection, in no
particular order
```
Hence `addAll()` may not be safe for preserving the order when adding to
the list. But since we are moving back to `TreeSet`, we can use `addAll()`
again.
---
[GitHub] incubator-tephra pull request #62: TEPHRA-245 Improve prune debug tool
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/62#discussion_r140183636
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/txprune/InvalidListPruningDebugTool.java
---
@@ -165,10 +165,10 @@ public void destroy() throws IOException {
*/
@Override
@SuppressWarnings("WeakerAccess")
- public SortedSet getIdleRegions(Integer
numRegions, String time) throws IOException {
+ public List getIdleRegions(Integer numRegions,
String time) throws IOException {
--- End diff --
ð
---
[GitHub] incubator-tephra issue #62: TEPHRA-245 Improve prune debug tool
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/62 @anew I have ported the changes to other compat modules. Please take a look. ---
[GitHub] incubator-tephra pull request #62: TEPHRA-245 Improve prune debug tool
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/62#discussion_r139777982
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/txprune/InvalidListPruningDebug.java
---
@@ -56,15 +59,18 @@
public class InvalidListPruningDebug {
private static final Logger LOG =
LoggerFactory.getLogger(InvalidListPruningDebug.class);
private static final Gson GSON = new Gson();
--- End diff --
Sure - will add that
---
[GitHub] incubator-tephra pull request #62: TEPHRA-245 Improve prune debug tool
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/62 TEPHRA-245 Improve prune debug tool JIRA - https://issues.apache.org/jira/browse/TEPHRA-245 Changes: 1. Improve documentation 2. Accept relative time as parameter 3. Print human readable dates 4. Sort the idle regions by prune upper bound 5. Fix warnings 6. Add test case You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/prune-debug-tool Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/62.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #62 commit f62d6a2b490c706c1e6280a70cc5b3155eda239d Author: poorna <poo...@cask.co> Date: 2017-09-14T00:08:29Z TEPHRA-245 Add TimeMathParser to parse relative time commit 6805f42f9c8ff25989e583e29a84fe33fd5ec051 Author: poorna <poo...@cask.co> Date: 2017-09-15T04:57:37Z TEPHRA-245 Improve prune debug tool 1. Improve documentation 2. Accept relative time as parameter 3. Print human readable dates 4. Sort the idle regions by prune upper bound 5. Fix warnings 6. Add test case ---
[GitHub] incubator-tephra pull request #61: TEPHRA-263 Enforce TTL, regardless of any...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/61#discussion_r139523472
--- Diff:
tephra-hbase-compat-1.0/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionVisibilityFilter.java
---
@@ -89,13 +89,13 @@ public TransactionVisibilityFilter(Transaction tx,
Map<byte[], Long> ttlByFamily
* {@link Filter.ReturnCode#INCLUDE_AND_NEXT_COL} will
be returned instead.
*/
public TransactionVisibilityFilter(Transaction tx, Map<byte[], Long>
ttlByFamily, boolean allowEmptyValues,
- ScanType scanType, @Nullable Filter
cellFilter) {
+ ScanType scanType, @Nullable Filter
cellFilter) {
this.tx = tx;
this.oldestTsByFamily = Maps.newTreeMap();
for (Map.Entry<byte[], Long> ttlEntry : ttlByFamily.entrySet()) {
long familyTTL = ttlEntry.getValue();
oldestTsByFamily.put(new ImmutableBytesWritable(ttlEntry.getKey()),
- familyTTL <= 0 ? 0 :
tx.getVisibilityUpperBound() - familyTTL * TxConstants.MAX_TX_PER_MS);
+ familyTTL <= 0 ? 0 : tx.getTransactionId() -
familyTTL * TxConstants.MAX_TX_PER_MS);
--- End diff --
We can have -ve timestamps here too
---
[GitHub] incubator-tephra issue #58: TEPHRA-231 Fix LICENSE and NOTICE files for rele...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/58 I did a build and was able to see the new license and notice files in the distribution jar. Can we rename `LICENSE_RELEASE` to `LICENSE_DISTRIBUTION` and `NOTICE_RELEASE` to `NOTICE_DISTRIBUTION`, since they only apply to the distribution tarball? Also is it okay to move them into `tephra-distribution/` directory for the same reason? In that case we can just call them `LICENSE` and `NOTICE` ---
[GitHub] incubator-tephra pull request #53: (TEPHRA-243) Improve logging for slow log...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/53#discussion_r138723289
--- Diff:
tephra-core/src/main/java/org/apache/tephra/persist/AbstractTransactionLog.java
---
@@ -85,44 +99,20 @@ public long getTimestamp() {
@Override
public void append(TransactionEdit edit) throws IOException {
-long startTime = System.nanoTime();
-synchronized (this) {
- ensureAvailable();
-
- Entry entry = new Entry(new
LongWritable(logSequence.getAndIncrement()), edit);
-
- // add to pending edits
- append(entry);
-}
-
-// wait for sync to complete
-sync();
-long durationMillis = (System.nanoTime() - startTime) / 100L;
-if (durationMillis > SLOW_APPEND_THRESHOLD) {
- LOG.info("Slow append to log " + getName() + ", took " +
durationMillis + " msec.");
-}
+append(Collections.singletonList(edit));
}
@Override
public void append(List edits) throws IOException {
-long startTime = System.nanoTime();
-synchronized (this) {
+// synchronizing here ensures that elements in the queue are ordered
by seq number
+synchronized (logSequence) {
ensureAvailable();
--- End diff --
`ensureAvailable()` can move out of the synchronized block since it uses a
couple of volatile variables and calls a synchronized method `init()`
---
[GitHub] incubator-tephra pull request #53: (TEPHRA-243) Improve logging for slow log...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/53#discussion_r138720628
--- Diff:
tephra-core/src/main/java/org/apache/tephra/persist/AbstractTransactionLog.java
---
@@ -134,57 +124,112 @@ private void ensureAvailable() throws IOException {
}
}
- /*
- * Appends new writes to the pendingWrites. It is better to keep it in
- * our own queue rather than writing it to the HDFS output stream because
- * HDFSOutputStream.writeChunk is not lightweight at all.
+ /**
+ * Return all pending writes at the time the method is called, or null
if no writes are pending.
+ *
+ * Note that after this method returns, there can be additional pending
writes,
+ * added concurrently while the existing pending writes are removed.
+ */
+ @Nullable
+ private Entry[] getPendingWrites() {
+synchronized (this) {
+ if (pendingWrites.isEmpty()) {
+return null;
+ }
+ Entry[] entriesToSync = new Entry[pendingWrites.size()];
+ for (int i = 0; i < entriesToSync.length; i++) {
+entriesToSync[i] = pendingWrites.remove();
+ }
+ return entriesToSync;
+}
+ }
+
+ /**
+ * When multiple threads try to log edits at the same time, they all
will call (@link #append}
+ * followed by {@link #sync()}, concurrently. Hence, it can happen that
multiple {@code append()}
+ * are followed by a single {@code sync}, or vice versa.
+ *
+ * We want to record the time and position of the first {@code append()}
after a {@code sync()},
+ * then measure the time after the next {@code sync()}, and log a
warning if it exceeds a threshold.
+ * Therefore this is called every time before we write the pending list
out to the log writer.
+ *
+ * See {@link #stopTimerIfNeeded(TransactionLogWriter)}.
+ *
+ * @throws IOException if the position of the writer cannot be determined
*/
- private void append(Entry e) throws IOException {
-pendingWrites.add(e);
+ private void startTimerIfNeeded(TransactionLogWriter writer, int
entryCount) throws IOException {
+// no sync needed because this is only called within a sync block
+if (positionBeforeWrite == -1L) {
+ positionBeforeWrite = writer.getPosition();
+ countSinceLastSync = 0;
+ stopWatch.reset().start();
+}
+countSinceLastSync += entryCount;
}
- // Returns all currently pending writes. New writes
- // will accumulate in a new list.
- private List getPendingWrites() {
-synchronized (this) {
- List save = this.pendingWrites;
- this.pendingWrites = new LinkedList<>();
- return save;
+ /**
+ * Called by a {@code sync()} after flushing to file system. Issues a
warning if the write(s)+sync
+ * together exceed a threshold.
+ *
+ * See {@link #startTimerIfNeeded(TransactionLogWriter, int)}.
+ *
+ * @throws IOException if the position of the writer cannot be determined
+ */
+ private void stopTimerIfNeeded(TransactionLogWriter writer) throws
IOException {
+// this method is only called by a thread if it actually called
sync(), inside a sync block
+if (positionBeforeWrite != -1L) { // actually it should never be -1,
but just in case
+ stopWatch.stop();
+ long elapsed = stopWatch.elapsedMillis();
+ if (elapsed >= slowAppendThreshold) {
+long currentPosition = writer.getPosition();
+long bytesWritten = currentPosition - positionBeforeWrite;
+LOG.info("Slow append to log {}, took {} ms for {} entr{} and {}
bytes.",
+ getName(), elapsed, countSinceLastSync,
countSinceLastSync == 1 ? "y" : "ies", bytesWritten);
+ }
}
+positionBeforeWrite = -1L;
+countSinceLastSync = 0;
}
private void sync() throws IOException {
// writes out pending entries to the HLog
-TransactionLogWriter tmpWriter = null;
long latestSeq = 0;
int entryCount = 0;
synchronized (this) {
if (closed) {
return;
}
- // prevent writer being dereferenced
- tmpWriter = writer;
-
- List currentPending = getPendingWrites();
- if (!currentPending.isEmpty()) {
-tmpWriter.commitMarker(currentPending.size());
- }
-
- // write out all accumulated entries to log.
- for (Entry e : currentPending) {
-tmpWriter.append(e);
-entryCount++;
-latestSeq = Math.max(latestSeq, e.getKey().get());
+ Entry[] currentPending = getPendingWrite
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/47#discussion_r138480498
--- Diff: tephra-core/src/main/thrift/transaction.thrift ---
@@ -73,15 +79,21 @@ service TTransactionServer {
// TODO remove this as it was replaced with startShortWithTimeout in 0.10
TTransaction startShortTimeout(1: i32 timeout),
TTransaction startShortClientId(1: string clientId) throws (1:
TGenericException e),
- TTransaction startShortWithClientIdAndTimeOut(1: string clientId, 2: i32
timeout) throws (1:TGenericException e),
- TTransaction startShortWithTimeout(1: i32 timeout) throws
(1:TGenericException e),
- TBoolean canCommitTx(1: TTransaction tx, 2: set changes) throws
(1:TTransactionNotInProgressException e),
- TBoolean canCommitOrThrow(1: TTransaction tx, 2: set changes)
throws (1:TTransactionNotInProgressException e,
-
2:TGenericException g,),
+ TTransaction startShortWithClientIdAndTimeOut(1: string clientId, 2: i32
timeout) throws (1: TGenericException e),
+ TTransaction startShortWithTimeout(1: i32 timeout) throws (1:
TGenericException e),
+ // TODO remove this as it was replaced with canCommitOrThrow in 0.13
+ TBoolean canCommitTx(1: TTransaction tx, 2: set changes) throws
(1: TTransactionNotInProgressException e),
+ void canCommitOrThrow(1: i64 tx, 2: set changes) throws (1:
TTransactionNotInProgressException e,
+ 2:
TTransactionConflictException c,
+ 3:
TGenericException g),
+ // TODO remove this as it was replaced with commitWithExn in 0.13
TBoolean commitTx(1: TTransaction tx) throws
(1:TTransactionNotInProgressException e),
+ void commitOrThrow(1: i64 txId, 2: i64 wp) throws (1:
TTransactionNotInProgressException e,
--- End diff --
I missed it yesterday, it would be good to use TransactionFailureException
here too for future-proofing.
---
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/47#discussion_r138477861
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -206,12 +210,19 @@ public TransactionManager(Configuration conf,
@Nonnull TransactionStateStorage p
// TODO: REMOVE WITH txnBackwardsCompatCheck()
longTimeoutTolerance = conf.getLong("data.tx.long.timeout.tolerance",
1);
-//
+ClientIdRetention retention = ClientIdRetention.valueOf(
--- End diff --
It would be good to catch the exception during enum conversion and use the
default value. This way transaction manager will still startup on a
misconfiguration.
---
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/47#discussion_r138238770
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionContext.java ---
@@ -311,25 +304,16 @@ private void persist() throws
TransactionFailureException {
}
private void commit() throws TransactionFailureException {
-boolean commitSuccess = false;
try {
- commitSuccess = txClient.commit(currentTx);
-} catch (TransactionNotInProgressException e) {
- String message = String.format("Transaction %d is not in progress.",
currentTx.getTransactionId());
- LOG.warn(message, e);
- abort(new TransactionFailureException(message, e));
- // abort will throw that exception
+ txClient.commitOrThrow(currentTx);
+} catch (TransactionNotInProgressException |
TransactionConflictException e) {
--- End diff --
This should also catch `TransactionFailureException` like the catch block
of method `checkForConflicts()` above, right?
---
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/47#discussion_r138239467 --- Diff: tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java --- @@ -89,21 +89,38 @@ * * @param tx transaction to verify * @param changeIds ids of changes made by transaction - * @return true if transaction can be committed otherwise false - * @throws TransactionSizeException if the size of the chgange set exceeds the allowed limit - * @throws TransactionNotInProgressException if the transaction is not in progress; most likely it has timed out. + * + * @throws TransactionSizeException if the size of the change set exceeds the allowed limit + * @throws TransactionConflictException if the change set has a conflict with an overlapping transaction + * @throws TransactionNotInProgressException if the transaction is not in progress; most likely it has timed out */ - boolean canCommitOrThrow(Transaction tx, Collection<byte[]> changeIds) throws TransactionFailureException; + void canCommitOrThrow(Transaction tx, Collection<byte[]> changeIds) +throws TransactionNotInProgressException, TransactionConflictException, TransactionSizeException; --- End diff -- Would declaring `TransactionFailureException` here help in making the API resilient to some future changes - in-case we need to add new exceptions to indicate a transaction failure? ---
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/47#discussion_r138237949
--- Diff:
tephra-api/src/main/java/org/apache/tephra/TransactionConflictException.java ---
@@ -22,11 +22,50 @@
* Thrown to indicate transaction conflict occurred when trying to commit
a transaction.
*/
public class TransactionConflictException extends
TransactionFailureException {
+
+ /**
+ * @deprecated since 0.13-incubating. Use {@link
#TransactionConflictException(long, String, String)} instead.
+ */
+ @Deprecated
public TransactionConflictException(String message) {
super(message);
+transactionId = null;
+conflictingChange = null;
+conflictingClient = null;
}
+ /**
+ * @deprecated since 0.13-incubating. Use {@link
#TransactionConflictException(long, String, String)} instead.
+ */
+ @Deprecated
public TransactionConflictException(String message, Throwable cause) {
super(message, cause);
+transactionId = null;
+conflictingChange = null;
+conflictingClient = null;
+ }
+
+ public TransactionConflictException(long transactionId, String
conflictingChange, String conflictingClient) {
+super(String.format("Transaction %d conflicts with %s on change key
'%s'", transactionId,
+conflictingClient == null ? "unknown client" :
conflictingClient, conflictingChange));
+this.transactionId = transactionId;
+this.conflictingChange = conflictingChange;
+this.conflictingClient = conflictingClient;
+ }
+
+ private final Long transactionId;
--- End diff --
It would be good if the fields are defined before the constructor
definitions.
---
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/47#discussion_r138270921
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -853,46 +867,45 @@ private void advanceWritePointer(long writePointer) {
}
}
- public boolean canCommit(Transaction tx, Collection<byte[]> changeIds)
-throws TransactionNotInProgressException, TransactionSizeException {
+ public void canCommit(long txId, Collection<byte[]> changeIds)
+throws TransactionNotInProgressException, TransactionSizeException,
TransactionConflictException {
txMetricsCollector.rate("canCommit");
Stopwatch timer = new Stopwatch().start();
-InProgressTx inProgressTx = inProgress.get(tx.getTransactionId());
+InProgressTx inProgressTx = inProgress.get(txId);
if (inProgressTx == null) {
synchronized (this) {
// invalid transaction, either this has timed out and moved to
invalid, or something else is wrong.
-if (invalidTxList.contains(tx.getTransactionId())) {
+if (invalidTxList.contains(txId)) {
throw new TransactionNotInProgressException(
String.format(
- "canCommit() is called for transaction %d that is not in
progress (it is known to be invalid)",
- tx.getTransactionId()));
+ "canCommit() is called for transaction %d that is not in
progress (it is known to be invalid)", txId));
} else {
throw new TransactionNotInProgressException(
-String.format("canCommit() is called for transaction %d that
is not in progress", tx.getTransactionId()));
+String.format("canCommit() is called for transaction %d that
is not in progress", txId));
}
}
}
Set set =
- validateChangeSet(tx, changeIds, inProgressTx.clientId != null ?
inProgressTx.clientId : DEFAULT_CLIENTID);
-
-if (hasConflicts(tx, set)) {
- return false;
+ validateChangeSet(txId, changeIds, inProgressTx.clientId != null ?
inProgressTx.clientId : DEFAULT_CLIENTID);
+for (byte[] change : changeIds) {
--- End diff --
I don't think this for-loop is needed as method `validateChangeSet()`
already returns `Set`.
---
[GitHub] incubator-tephra pull request #47: [TEPHRA-240] Include conflicting key and ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/47#discussion_r138222710
--- Diff:
tephra-api/src/main/java/org/apache/tephra/TransactionConflictException.java ---
@@ -22,11 +22,50 @@
* Thrown to indicate transaction conflict occurred when trying to commit
a transaction.
*/
public class TransactionConflictException extends
TransactionFailureException {
+
+ /**
+ * @deprecated since 0.13-incubating. Use {@link
#TransactionConflictException(long, String, String)} instead.
+ */
+ @Deprecated
public TransactionConflictException(String message) {
super(message);
+transactionId = null;
+conflictingChange = null;
+conflictingClient = null;
}
+ /**
+ * @deprecated since 0.13-incubating. Use {@link
#TransactionConflictException(long, String, String)} instead.
+ */
+ @Deprecated
public TransactionConflictException(String message, Throwable cause) {
super(message, cause);
+transactionId = null;
+conflictingChange = null;
+conflictingClient = null;
+ }
+
+ public TransactionConflictException(long transactionId, String
conflictingChange, String conflictingClient) {
--- End diff --
I think `conflictingKey` would be a more appropriate name for
`conflictingChange`.
---
[GitHub] incubator-tephra pull request #55: TEPHRA-244 Remove regions of deleted tabl...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/55#discussion_r137981943
--- Diff: tephra-hbase-compat-1.1-base/src/test/resources/logback-test.xml
---
@@ -0,0 +1,39 @@
+
+
+
+
+
+
+ %d{ISO8601} - %-5p [%t:%C{1}@%L] - %m%n
+
+
+
+
+
--- End diff --
I'll copy over this file to the other compat modules
---
[GitHub] incubator-tephra pull request #55: TEPHRA-244 Remove regions of deleted tabl...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/55#discussion_r137981906
--- Diff:
tephra-hbase-compat-1.1-base/src/test/java/org/apache/tephra/hbase/txprune/InvalidListPruneTest.java
---
@@ -384,7 +384,67 @@ public void testPruneEmptyTable() throws Exception {
hBaseAdmin.disableTable(txEmptyTable);
hBaseAdmin.deleteTable(txEmptyTable);
}
+ }
+
+ @Test
+ public void testPruneTransientTable() throws Exception {
+// Make sure that transient tables do not block the progress of pruning
+
+// Create a temp table
+TableName txTempTable = TableName.valueOf("tempTable");
+createTable(txTempTable.getName(), new byte[][]{family}, false,
+
Collections.singletonList(TestTransactionProcessor.class.getName()));
+
+TableName txDataTable2 = null;
+
+TransactionPruningPlugin transactionPruningPlugin = new
TestTransactionPruningPlugin();
+transactionPruningPlugin.initialize(conf);
+
+try {
+ long now1 = System.currentTimeMillis();
+ long inactiveTxTimeNow1 = (now1 - 150) * TxConstants.MAX_TX_PER_MS;
+ long noPruneUpperBound = -1;
+ long expectedPruneUpperBound1 = (now1 - 200) *
TxConstants.MAX_TX_PER_MS;
+ InMemoryTransactionStateCache.setTransactionSnapshot(
+new TransactionSnapshot(expectedPruneUpperBound1,
expectedPruneUpperBound1, expectedPruneUpperBound1,
+ImmutableSet.of(expectedPruneUpperBound1),
+ImmutableSortedMap.<Long,
TransactionManager.InProgressTx>of()));
+
+ // fetch prune upper bound, there should be no prune upper bound
since nothing has been compacted yet.
+ // This run is only to store the initial set of regions
+ long pruneUpperBound1 =
transactionPruningPlugin.fetchPruneUpperBound(now1, inactiveTxTimeNow1);
+ Assert.assertEquals(noPruneUpperBound, pruneUpperBound1);
+ transactionPruningPlugin.pruneComplete(now1, noPruneUpperBound);
+
+ // Now delete the transient table
+ hBaseAdmin.disableTable(txTempTable);
+ hBaseAdmin.deleteTable(txTempTable);
+
+ // Compact the data table now
+ testUtil.compact(txDataTable1, true);
+ // Since the write to prune table happens async, we need to sleep a
bit before checking the state of the table
+ TimeUnit.SECONDS.sleep(2);
+
+ // Create a new table that will not be compacted
+ txDataTable2 = TableName.valueOf("invalidListPruneTestTable2");
+ createTable(txDataTable2.getName(), new byte[][]{family}, false,
+
Collections.singletonList(TestTransactionProcessor.class.getName()));
+
+ // fetch prune upper bound, there should be a prune upper bound even
though txTempTable does not exist anymore,
+ // and txDataTable2 has not been compacted/flushed yet
+ long now2 = System.currentTimeMillis();
+ long inactiveTxTimeNow2 = (now1 - 150) * TxConstants.MAX_TX_PER_MS;
+ long pruneUpperBound2 =
transactionPruningPlugin.fetchPruneUpperBound(now2, inactiveTxTimeNow2);
+ Assert.assertEquals(expectedPruneUpperBound1, pruneUpperBound2);
+ transactionPruningPlugin.pruneComplete(now2,
expectedPruneUpperBound1);
+} finally {
+ transactionPruningPlugin.destroy();
+ if (txDataTable2 != null) {
--- End diff --
ð
---
[GitHub] incubator-tephra pull request #49: [TEPHRA-242] Ensure Pruning Service shuts...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/49#discussion_r137865504
--- Diff:
tephra-core/src/main/java/org/apache/tephra/distributed/TransactionService.java
---
@@ -121,15 +121,15 @@ public void failed(State from, Throwable failure) {
@Override
public void follower() {
ListenableFuture stopFuture = null;
+if (pruningService != null && pruningService.isRunning()) {
+ // Wait for pruning service to stop after un-registering from
discovery
+ stopFuture = pruningService.stop();
+}
// First stop the transaction server as un-registering from
discovery can block sometimes.
// That can lead to multiple transaction servers being active at
the same time.
if (server != null && server.isRunning()) {
server.stopAndWait();
}
-if (pruningService != null && pruningService.isRunning()) {
- // Wait for pruning service to stop after un-registering from
discovery
- stopFuture = pruningService.stop();
-}
undoRegister();
if (stopFuture != null) {
--- End diff --
It would be good to add a timeout to the `Futures.getUnchecked(stopFuture)`
in the next line, so that we don't block forever.
---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r131594276
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -177,11 +173,23 @@ public void close() throws IOException {
} catch (IOException e) {
try {
secondaryIndexTable.close();
+conn.close();
} catch (IOException ex) {
e.addSuppressed(e);
}
throw e;
}
secondaryIndexTable.close();
+conn.close();
+ }
+
+ protected void finalize() throws Throwable {
--- End diff --
We try not to rely on `finalize()` since there is no guarantee when it will
be called. Also since we are closing the connection in the `close()` method,
there is no need for finalize. Let's remove this method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r131594887
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -177,11 +173,23 @@ public void close() throws IOException {
} catch (IOException e) {
try {
secondaryIndexTable.close();
+conn.close();
--- End diff --
`conn` may not be closed if `secondaryIndexTable.close()` throws an
exception.
Adding another nested try-catch will make the code complex. Instead what do
you think about putting each close in its own try-catch block, remember the
first exception, and add the other exceptions as suppressed to the first
exception?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra issue #45: [TEPHRA-236] Changes to replace deprecated class...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/45 @bijugs Just one more comment to address - https://github.com/apache/incubator-tephra/pull/45#discussion_r128894170 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r128894190
--- Diff:
tephra-hbase-compat-1.1-base/src/test/java/org/apache/tephra/hbase/TransactionAwareHTableTest.java
---
@@ -180,6 +182,7 @@ public static void setupBeforeClass() throws Exception {
testUtil.startMiniCluster();
hBaseAdmin = testUtil.getHBaseAdmin();
+conn = testUtil.getConnection();
--- End diff --
ð
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r128894187
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -56,17 +58,24 @@
private static final byte[] secondaryIndexQualifier = Bytes.toBytes('r');
private static final byte[] DELIMITER = new byte[] {0};
- public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, HTableInterface hTable,
- byte[] secondaryIndex) {
+ public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, Table hTable,
+ byte[] secondaryIndex) throws IOException {
secondaryIndexTableName =
TableName.valueOf(hTable.getName().getNameAsString() + ".idx");
-HTable secondaryIndexHTable = null;
-try (HBaseAdmin hBaseAdmin = new
HBaseAdmin(hTable.getConfiguration())) {
+Table secondaryIndexHTable = null;
+Connection conn = null;
+try {
+ conn =
ConnectionFactory.createConnection(hTable.getConfiguration());
+ Admin hBaseAdmin = conn.getAdmin();
if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
hBaseAdmin.createTable(new
HTableDescriptor(secondaryIndexTableName));
}
- secondaryIndexHTable = new HTable(hTable.getConfiguration(),
secondaryIndexTableName);
-} catch (Exception e) {
+ secondaryIndexHTable = conn.getTable(secondaryIndexTableName);
+} catch (Exception e) {
--- End diff --
ð
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r128894170
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -56,17 +58,24 @@
private static final byte[] secondaryIndexQualifier = Bytes.toBytes('r');
private static final byte[] DELIMITER = new byte[] {0};
- public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, HTableInterface hTable,
- byte[] secondaryIndex) {
+ public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, Table hTable,
+ byte[] secondaryIndex) throws IOException {
secondaryIndexTableName =
TableName.valueOf(hTable.getName().getNameAsString() + ".idx");
-HTable secondaryIndexHTable = null;
-try (HBaseAdmin hBaseAdmin = new
HBaseAdmin(hTable.getConfiguration())) {
+Table secondaryIndexHTable = null;
+Connection conn = null;
+try {
+ conn =
ConnectionFactory.createConnection(hTable.getConfiguration());
+ Admin hBaseAdmin = conn.getAdmin();
if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
hBaseAdmin.createTable(new
HTableDescriptor(secondaryIndexTableName));
}
- secondaryIndexHTable = new HTable(hTable.getConfiguration(),
secondaryIndexTableName);
-} catch (Exception e) {
+ secondaryIndexHTable = conn.getTable(secondaryIndexTableName);
+} catch (Exception e) {
Throwables.propagate(e);
+} finally {
+ if (conn != null) {
+ conn.close();
--- End diff --
@bijugs The connection created in the constructor has to be closed after
the table is closed. Take a look at
https://hbase.apache.org/1.1/apidocs/org/apache/hadoop/hbase/client/ConnectionFactory.html
for an example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r126789926
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -56,17 +58,24 @@
private static final byte[] secondaryIndexQualifier = Bytes.toBytes('r');
private static final byte[] DELIMITER = new byte[] {0};
- public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, HTableInterface hTable,
- byte[] secondaryIndex) {
+ public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, Table hTable,
+ byte[] secondaryIndex) throws IOException {
secondaryIndexTableName =
TableName.valueOf(hTable.getName().getNameAsString() + ".idx");
-HTable secondaryIndexHTable = null;
-try (HBaseAdmin hBaseAdmin = new
HBaseAdmin(hTable.getConfiguration())) {
+Table secondaryIndexHTable = null;
+Connection conn = null;
+try {
+ conn =
ConnectionFactory.createConnection(hTable.getConfiguration());
+ Admin hBaseAdmin = conn.getAdmin();
if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
hBaseAdmin.createTable(new
HTableDescriptor(secondaryIndexTableName));
}
- secondaryIndexHTable = new HTable(hTable.getConfiguration(),
secondaryIndexTableName);
-} catch (Exception e) {
+ secondaryIndexHTable = conn.getTable(secondaryIndexTableName);
+} catch (Exception e) {
--- End diff --
Extra space at the end of the line causes the following warning when the
[patch](https://github.com/apache/incubator-tephra/pull/45.patch) for this PR
is applied -
```
45.patch:49: trailing whitespace.
try {
45.patch:58: trailing whitespace.
} catch (Exception e) {
45.patch:298: trailing whitespace.
warning: 3 lines add whitespace errors.
```
Can you remove the extra spaces?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #45: [TEPHRA-236] Changes to replace deprecate...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/45#discussion_r126789026
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/SecondaryIndexTable.java
---
@@ -56,17 +58,24 @@
private static final byte[] secondaryIndexQualifier = Bytes.toBytes('r');
private static final byte[] DELIMITER = new byte[] {0};
- public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, HTableInterface hTable,
- byte[] secondaryIndex) {
+ public SecondaryIndexTable(TransactionServiceClient
transactionServiceClient, Table hTable,
+ byte[] secondaryIndex) throws IOException {
secondaryIndexTableName =
TableName.valueOf(hTable.getName().getNameAsString() + ".idx");
-HTable secondaryIndexHTable = null;
-try (HBaseAdmin hBaseAdmin = new
HBaseAdmin(hTable.getConfiguration())) {
+Table secondaryIndexHTable = null;
+Connection conn = null;
+try {
+ conn =
ConnectionFactory.createConnection(hTable.getConfiguration());
+ Admin hBaseAdmin = conn.getAdmin();
if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
hBaseAdmin.createTable(new
HTableDescriptor(secondaryIndexTableName));
}
- secondaryIndexHTable = new HTable(hTable.getConfiguration(),
secondaryIndexTableName);
-} catch (Exception e) {
+ secondaryIndexHTable = conn.getTable(secondaryIndexTableName);
+} catch (Exception e) {
Throwables.propagate(e);
+} finally {
+ if (conn != null) {
+ conn.close();
--- End diff --
The connection has to be closed in the method `SecondaryIndexTable.close()`
too.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra issue #45: [TEPHRA-236] Changes to replace deprecated class...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/45 Thanks @bijugs, I'll continue the review --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #44: TEPHRA-235 Ensure that TransactionSnapsho...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/44#discussion_r123363497
--- Diff:
tephra-core/src/test/java/org/apache/tephra/snapshot/SnapshotCodecTest.java ---
@@ -375,4 +386,17 @@ private void
assertTransactionVisibilityStateEquals(TransactionVisibilityState e
Assert.assertEquals(expected.getInProgress(), input.getInProgress());
Assert.assertEquals(expected.getInvalid(), input.getInvalid());
}
+
+ private TransactionSnapshot assertSorted(TransactionSnapshot txSnapshot)
{
+Collection invalidTxList = txSnapshot.getInvalid();
+if (invalidTxList.isEmpty()) {
+ return txSnapshot;
+}
+Long previousLong = Iterables.get(invalidTxList, 0);
+for (Long aLong : invalidTxList) {
+ Assert.assertTrue(aLong >= previousLong);
--- End diff --
Would be good to say sort error here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117851160
--- Diff:
tephra-core/src/main/java/org/apache/tephra/distributed/TransactionServiceClient.java
---
@@ -125,15 +130,20 @@ public static void doMain(boolean verbose,
Configuration conf) throws Exception
}
}
+ public TransactionServiceClient(Configuration config,
ThriftClientProvider clientProvider) {
--- End diff --
Please add javadocs for the public constructor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117841761
--- Diff:
tephra-core/src/main/java/org/apache/tephra/runtime/TransactionDistributedModule.java
---
@@ -35,13 +34,24 @@
import org.apache.tephra.persist.TransactionStateStorage;
import org.apache.tephra.snapshot.SnapshotCodecProvider;
+import java.lang.management.ManagementFactory;
+
/**
* Guice bindings for running in distributed mode on a cluster.
*/
-final class TransactionDistributedModule extends AbstractModule {
+final class TransactionDistributedModule extends
ClientIdAwareTransactionModule {
+
+ public TransactionDistributedModule() {
+super(ManagementFactory.getRuntimeMXBean().getName());
--- End diff --
In that case they can provide the client id since the default constructor
won't be present, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117834468
--- Diff:
tephra-core/src/main/java/org/apache/tephra/runtime/TransactionDistributedModule.java
---
@@ -35,13 +34,24 @@
import org.apache.tephra.persist.TransactionStateStorage;
import org.apache.tephra.snapshot.SnapshotCodecProvider;
+import java.lang.management.ManagementFactory;
+
/**
* Guice bindings for running in distributed mode on a cluster.
*/
-final class TransactionDistributedModule extends AbstractModule {
+final class TransactionDistributedModule extends
ClientIdAwareTransactionModule {
+
+ public TransactionDistributedModule() {
+super(ManagementFactory.getRuntimeMXBean().getName());
--- End diff --
This default is already defined in `TransactionModules`. Do we still need
it here?
Same for `TransactionInMemoryModule` and `TransactionLocalModule`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117552592
--- Diff:
tephra-hbase-compat-0.98/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java
---
@@ -127,7 +126,7 @@ public TransactionProcessor() {
public void start(CoprocessorEnvironment e) throws IOException {
if (e instanceof RegionCoprocessorEnvironment) {
RegionCoprocessorEnvironment env = (RegionCoprocessorEnvironment) e;
- Supplier cacheSupplier =
getTransactionStateCacheSupplier(env);
+ this.cacheSupplier = getTransactionStateCacheSupplier(env);
--- End diff --
Unrelated change?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117552141
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1378,17 +1403,32 @@ public TransactionType getTransactionType() {
private final long expiration;
private final InProgressType type;
private final LongArrayList checkpointWritePointers;
+private final String clientId;
--- End diff --
It is okay not to add it now since we are not persisting it. Add a comment
in the code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117338264
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -763,15 +781,22 @@ private long getNextWritePointer() {
* transaction moves it to the invalid list because we assume that its
writes cannot be rolled back.
*/
public Transaction startLong() {
+return startLong(null);
+ }
+
+ /**
+ * Starts a long transaction with a client id.
+ */
+ public Transaction startLong(String clientId) {
--- End diff --
Eventually we want all client's to pass in a client i, hence the
deprecation. If we allow nulls then that defeats the purpose, no?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117129143
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -763,15 +781,22 @@ private long getNextWritePointer() {
* transaction moves it to the invalid list because we assume that its
writes cannot be rolled back.
*/
public Transaction startLong() {
+return startLong(null);
+ }
+
+ /**
+ * Starts a long transaction with a client id.
+ */
+ public Transaction startLong(String clientId) {
--- End diff --
Should we throw an exception if `clientId` is null?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/42#discussion_r117127048
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -347,15 +346,17 @@ private void cleanupTimedOutTransactions() {
long currentTime = System.currentTimeMillis();
Map<Long, InProgressType> timedOut = Maps.newHashMap();
for (Map.Entry<Long, InProgressTx> tx : inProgress.entrySet()) {
- long expiration = tx.getValue().getExpiration();
+ InProgressTx inProgressTx = tx.getValue();
+ long expiration = inProgressTx.getExpiration();
if (expiration >= 0L && currentTime > expiration) {
// timed out, remember tx id (can't remove while iterating
over entries)
-timedOut.put(tx.getKey(), tx.getValue().getType());
-LOG.info("Tx invalid list: added tx {} because of timeout",
tx.getKey());
+timedOut.put(tx.getKey(), inProgressTx.getType());
+String clientId = inProgressTx.getClientId() != null ?
inProgressTx.getClientId() : "unknown";
+LOG.info("Tx invalid list: added tx {} belonging to client
'{}' because of timeout.",
--- End diff --
Let's add the client id to the log message in `doInvalidate` too.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/42#discussion_r117121506 --- Diff: tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java --- @@ -44,11 +44,33 @@ Transaction startShort(int timeout); /** + * Starts a new short transaction. + * @param clientId id of the client --- End diff -- Let's be a little more descriptive, something like - `An id that can be used to identify the client starting the transaction` --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #42: TEPHRA-228 Adding the ability to pass-in ...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/42#discussion_r117117388 --- Diff: tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java --- @@ -44,11 +44,33 @@ Transaction startShort(int timeout); --- End diff -- It would be good to deprecate all the start methods that do not take in a client id. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #43: TEPHRA-230 Fix reading of saved regions d...
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/43 TEPHRA-230 Fix reading of saved regions during invalid pruning run JIRA - https://issues.apache.org/jira/browse/TEPHRA-230 You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/fix-invalid-prune Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/43.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #43 commit 667144a99cd3863bf68982fb745d15786f294a92 Author: poorna <poo...@cask.co> Date: 2017-05-17T00:47:15Z Port to other compat modules commit daf57aa167c0c6c240ba04642d0c328211cb1b20 Author: poorna <poo...@cask.co> Date: 2017-05-17T00:51:32Z TEPHRA-230 Fix reading of saved regions during invalid pruning run --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra issue #41: TEPHRA-152 Using ReferenceCounting for Transacti...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/41 @gokulavasan Now that we have introduced class `ReferenceCountedSupplier`, it would be good to use this class in `PruneUpperBoundWriterSupplier` too. It will reduce code duplication. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #38: TEPHRA-224 Handle delay between transacti...
Github user poornachandra closed the pull request at: https://github.com/apache/incubator-tephra/pull/38 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra issue #38: TEPHRA-224 Handle delay between transaction max ...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/38 Merged this PR to branch release/0.11.0-incubating --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra issue #38: TEPHRA-224 Handle delay between transaction max ...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/38 @anew @chtyim I have addressed comments. Please take another look. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #38: TEPHRA-224 Handle delay between transacti...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/38#discussion_r102559503
--- Diff:
tephra-core/src/main/java/org/apache/tephra/txprune/TransactionPruningRunnable.java
---
@@ -57,8 +60,13 @@ public void run() {
Transaction tx = txManager.startShort();
txManager.abort(tx);
+ if (tx.getInvalids().length == 0) {
+LOG.info("Invalid list is empty, not running transaction pruning");
--- End diff --
I have filed TEPHRA-225 for to handle this
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #38: TEPHRA-224 Handle delay between transacti...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/38#discussion_r102366609
--- Diff:
tephra-core/src/main/java/org/apache/tephra/txprune/TransactionPruningRunnable.java
---
@@ -57,8 +60,13 @@ public void run() {
Transaction tx = txManager.startShort();
txManager.abort(tx);
+ if (tx.getInvalids().length == 0) {
+LOG.info("Invalid list is empty, not running transaction pruning");
--- End diff --
Yes - I was planning on making that change later since that needs some
analysis. I'll file a JIRA for that. Is that okay?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #38: TEPHRA-224 Handle delay between transacti...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/38#discussion_r102366454 --- Diff: tephra-core/src/main/java/org/apache/tephra/TxConstants.java --- @@ -376,6 +376,11 @@ public static final String PRUNE_FLUSH_INTERVAL = "data.tx.prune.flush.interval"; /** + * The buffer in seconds used to pad transaction max lifetime while pruning. + */ +public static final String PRUNE_BUFFER_INTERVAL = "data.tx.prune.buffer"; --- End diff -- What would be a good name? `PRUNE_BUFFER_TIME`? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #38: TEPHRA-224 Handle delay between transacti...
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/38 TEPHRA-224 Handle delay between transaction max lifetime check and data writes while pruning JIRA - https://issues.apache.org/jira/browse/TEPHRA-224 You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/pruning-buffer Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/38.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #38 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #37: TEPHRA-223 Encapsulate the two data struc...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/37#discussion_r102348853
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1123,15 +1112,14 @@ private boolean doTruncateInvalidTxBefore(long
time) throws InvalidTruncateTimeE
}
// Find all invalid transactions earlier than truncateWp
-Set toTruncate = Sets.newHashSet();
-for (long wp : invalid) {
- // invalid list is sorted, hence can stop as soon as we reach a wp
>= truncateWp
- if (wp >= truncateWp) {
-break;
+LongSet toTruncate = new LongArraySet();
+for (long wp : invalidTxList.toRawList()) {
--- End diff --
Good point
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra issue #37: TEPHRA-223 Encapsulate the two data structures u...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/37 @anew @gokulavasan I have addressed the comments, please take another look --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #37: TEPHRA-223 Encapsulate the two data struc...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/37#discussion_r102127364
--- Diff:
tephra-core/src/main/java/org/apache/tephra/manager/InvalidTxList.java ---
@@ -0,0 +1,110 @@
+/*
+ * 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.tephra.manager;
+
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import org.apache.tephra.TransactionManager;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.concurrent.NotThreadSafe;
+
+/**
+ * This is an internal class used by the {@link TransactionManager} to
store invalid transaction ids.
+ * This class uses both a list and an array to keep track of the invalid
ids. The list is the primary
+ * data structure for storing the invalid ids. The array is populated
lazily on changes to the list.
+ * The array is used to avoid creating a new array every time method
{@link #toArray()} is invoked.
+ *
+ * This class is not thread safe and relies on external synchronization.
TransactionManager always
+ * accesses an instance of this class after synchronization.
+ */
+@NotThreadSafe
+public class InvalidTxList {
+ private static final long[] NO_INVALID_TX = { };
+
+ private final LongArrayList invalid = new LongArrayList();
+ private long[] invalidArray = NO_INVALID_TX;
+
+ private boolean dirty = false; // used to track changes to the invalid
list
+
+ public int size() {
+return invalid.size();
+ }
+
+ public boolean isEmpty() {
+return invalid.isEmpty();
+ }
+
+ public boolean add(long id) {
+boolean changed = invalid.add(id);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public boolean addAll(Collection ids) {
+boolean changed = invalid.addAll(ids);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public boolean contains(long id) {
+return invalid.contains(id);
+ }
+
+ public boolean remove(long id) {
+boolean changed = invalid.rem(id);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public boolean removeAll(Collection ids) {
+boolean changed = invalid.removeAll(ids);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public void clear() {
+invalid.clear();
+invalidArray = NO_INVALID_TX;
+dirty = false;
+ }
+
+ public long[] toArray() {
+if (dirty) {
+ Collections.sort(invalid);
+ invalidArray = invalid.toLongArray();
+ dirty = false;
+}
+return invalidArray;
+ }
+
+ public List toList() {
+return Collections.unmodifiableList(invalid);
--- End diff --
The sorting behavior is consistent with the existing implementation. Also
truncate and snapshot (the two places where toList() is used) are not invoked
frequently. So I suggest that, we'll leave this behavior as-is for now. We can
take a look at this again when working on the performance improvements.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #37: TEPHRA-223 Encapsulate the two data struc...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/37#discussion_r102125886
--- Diff:
tephra-core/src/main/java/org/apache/tephra/manager/InvalidTxList.java ---
@@ -0,0 +1,110 @@
+/*
+ * 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.tephra.manager;
+
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import org.apache.tephra.TransactionManager;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.concurrent.NotThreadSafe;
+
+/**
+ * This is an internal class used by the {@link TransactionManager} to
store invalid transaction ids.
+ * This class uses both a list and an array to keep track of the invalid
ids. The list is the primary
+ * data structure for storing the invalid ids. The array is populated
lazily on changes to the list.
+ * The array is used to avoid creating a new array every time method
{@link #toArray()} is invoked.
+ *
+ * This class is not thread safe and relies on external synchronization.
TransactionManager always
+ * accesses an instance of this class after synchronization.
+ */
+@NotThreadSafe
+public class InvalidTxList {
+ private static final long[] NO_INVALID_TX = { };
+
+ private final LongArrayList invalid = new LongArrayList();
+ private long[] invalidArray = NO_INVALID_TX;
+
+ private boolean dirty = false; // used to track changes to the invalid
list
+
+ public int size() {
+return invalid.size();
+ }
+
+ public boolean isEmpty() {
+return invalid.isEmpty();
+ }
+
+ public boolean add(long id) {
+boolean changed = invalid.add(id);
+if (changed && !dirty) {
--- End diff --
Thanks, it makes the code more readable
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #37: TEPHRA-223 Encapsulate the two data struc...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/37#discussion_r102123021
--- Diff:
tephra-core/src/main/java/org/apache/tephra/manager/InvalidTxList.java ---
@@ -0,0 +1,110 @@
+/*
+ * 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.tephra.manager;
+
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import org.apache.tephra.TransactionManager;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.concurrent.NotThreadSafe;
+
+/**
+ * This is an internal class used by the {@link TransactionManager} to
store invalid transaction ids.
+ * This class uses both a list and an array to keep track of the invalid
ids. The list is the primary
+ * data structure for storing the invalid ids. The array is populated
lazily on changes to the list.
+ * The array is used to avoid creating a new array every time method
{@link #toArray()} is invoked.
+ *
+ * This class is not thread safe and relies on external synchronization.
TransactionManager always
+ * accesses an instance of this class after synchronization.
+ */
+@NotThreadSafe
+public class InvalidTxList {
+ private static final long[] NO_INVALID_TX = { };
+
+ private final LongArrayList invalid = new LongArrayList();
+ private long[] invalidArray = NO_INVALID_TX;
+
+ private boolean dirty = false; // used to track changes to the invalid
list
+
+ public int size() {
+return invalid.size();
+ }
+
+ public boolean isEmpty() {
+return invalid.isEmpty();
+ }
+
+ public boolean add(long id) {
+boolean changed = invalid.add(id);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public boolean addAll(Collection ids) {
+boolean changed = invalid.addAll(ids);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public boolean contains(long id) {
+return invalid.contains(id);
+ }
+
+ public boolean remove(long id) {
+boolean changed = invalid.rem(id);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public boolean removeAll(Collection ids) {
+boolean changed = invalid.removeAll(ids);
+if (changed && !dirty) {
+ dirty = true;
+}
+return changed;
+ }
+
+ public void clear() {
+invalid.clear();
+invalidArray = NO_INVALID_TX;
+dirty = false;
+ }
+
+ public long[] toArray() {
+if (dirty) {
+ Collections.sort(invalid);
+ invalidArray = invalid.toLongArray();
+ dirty = false;
+}
+return invalidArray;
+ }
+
+ public List toList() {
+return Collections.unmodifiableList(invalid);
--- End diff --
Good catch. I'll update InvalidTxList to lazily sort both the list and the
array.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #37: TEPHRA-223 Encapsulate the two data struc...
GitHub user poornachandra opened a pull request: https://github.com/apache/incubator-tephra/pull/37 TEPHRA-223 Encapsulate the two data structures used for invalid transactions to avoid update issues JIRA - https://issues.apache.org/jira/browse/TEPHRA-223 Approach: Encapsulate the invalid list and the invalidArray data structures into a class InvalidTxList. InvalidTxList tracks the changes to invalid list, and then populates invalidArray lazily when the list changes. TODO: Add tests for class InvalidTxList You can merge this pull request into a Git repository by running: $ git pull https://github.com/poornachandra/incubator-tephra feature/invalid-persistence Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/37.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #37 commit 41f293e6f23a307a7709aba0c3a625abb29f9d4d Author: poorna <poo...@cask.co> Date: 2017-02-21T01:13:39Z TEPHRA-223 Encapsulate the two data structures used for invalid transactions to avoid update issues commit ff2996916307ee49c215aba9758b03ad7b318db9 Author: poorna <poo...@cask.co> Date: 2017-02-21T01:52:20Z TEPHRA-223 Synchronize the access to invalid list --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra issue #35: (TEPHRA-219) Execute cross region calls in Copro...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/35 LGTM --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100940812
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/txprune/PruneUpperBoundWriter.java
---
@@ -97,27 +101,33 @@ private void startFlushThread() {
flushThread = new Thread("tephra-prune-upper-bound-writer") {
@Override
public void run() {
-while ((!isInterrupted()) && isRunning()) {
+while ((!isInterrupted()) && (!stopped)) {
long now = System.currentTimeMillis();
if (now > (lastChecked + pruneFlushInterval)) {
// should flush data
try {
- // Record prune upper bound
- while (!pruneEntries.isEmpty()) {
-Map.Entry<byte[], Long> firstEntry =
pruneEntries.firstEntry();
-
dataJanitorState.savePruneUpperBoundForRegion(firstEntry.getKey(),
firstEntry.getValue());
-// We can now remove the entry only if the key and value
match with what we wrote since it is
-// possible that a new pruneUpperBound for the same key
has been added
-pruneEntries.remove(firstEntry.getKey(),
firstEntry.getValue());
- }
- // Record empty regions
- while (!emptyRegions.isEmpty()) {
-Map.Entry<byte[], Long> firstEntry =
emptyRegions.firstEntry();
-
dataJanitorState.saveEmptyRegionForTime(firstEntry.getValue(),
firstEntry.getKey());
-// We can now remove the entry only if the key and value
match with what we wrote since it is
-// possible that a new value for the same key has been
added
-emptyRegions.remove(firstEntry.getKey(),
firstEntry.getValue());
- }
+ User.runAsLoginUser(new PrivilegedExceptionAction() {
--- End diff --
Let's not close TEPHRA-219 until we have figured out how to handle the
wrapping cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100940705
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java
---
@@ -454,28 +442,38 @@ protected Filter getTransactionFilter(Transaction tx,
ScanType type, Filter filt
return TransactionFilters.getVisibilityFilter(tx, ttlByFamily,
allowEmptyValues, type, filter);
}
- private void
initPruneState(ObserverContext c) {
-Configuration conf = getConfiguration(c.getEnvironment());
-// Configuration won't be null in TransactionProcessor but the derived
classes might return
-// null if it is not available temporarily
+ protected void initializePruneState(RegionCoprocessorEnvironment env) {
--- End diff --
Good to add javadoc for the protected method
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100931804
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java
---
@@ -454,28 +441,39 @@ protected Filter getTransactionFilter(Transaction tx,
ScanType type, Filter filt
return TransactionFilters.getVisibilityFilter(tx, ttlByFamily,
allowEmptyValues, type, filter);
}
- private void
initPruneState(ObserverContext c) {
-Configuration conf = getConfiguration(c.getEnvironment());
-// Configuration won't be null in TransactionProcessor but the derived
classes might return
-// null if it is not available temporarily
+ protected void initializePruneState(RegionCoprocessorEnvironment env) {
+Configuration conf = getConfiguration(env);
if (conf != null) {
pruneEnable =
conf.getBoolean(TxConstants.TransactionPruning.PRUNE_ENABLE,
TxConstants.TransactionPruning.DEFAULT_PRUNE_ENABLE);
+
if (Boolean.TRUE.equals(pruneEnable)) {
-String pruneTable =
conf.get(TxConstants.TransactionPruning.PRUNE_STATE_TABLE,
-
TxConstants.TransactionPruning.DEFAULT_PRUNE_STATE_TABLE);
-long pruneFlushInterval = TimeUnit.SECONDS.toMillis(
- conf.getLong(TxConstants.TransactionPruning.PRUNE_FLUSH_INTERVAL,
-
TxConstants.TransactionPruning.DEFAULT_PRUNE_FLUSH_INTERVAL));
-compactionState = new CompactionState(c.getEnvironment(),
TableName.valueOf(pruneTable), pruneFlushInterval);
+// pruneTable and pruneFlushInterval cannot be changed by simply
loading the Configuration dynamically
+// since we have only one flush thread across all regions and we
might loose.
+TableName pruneTable =
TableName.valueOf(conf.get(TxConstants.TransactionPruning.PRUNE_STATE_TABLE,
+
TxConstants.TransactionPruning.DEFAULT_PRUNE_STATE_TABLE));
+long pruneFlushInterval = TimeUnit.SECONDS.toMillis(conf.getLong(
+ TxConstants.TransactionPruning.PRUNE_FLUSH_INTERVAL,
+ TxConstants.TransactionPruning.DEFAULT_PRUNE_FLUSH_INTERVAL));
+
+compactionState = new CompactionState(env, pruneTable,
pruneFlushInterval);
if (LOG.isDebugEnabled()) {
- LOG.debug("Automatic invalid list pruning is enabled. Compaction
state will be recorded in table "
- + pruneTable);
+ LOG.debug(String.format("Automatic invalid list pruning is
enabled for table %s. Compaction state " +
+"will be recorded in table %s",
+
env.getRegionInfo().getTable().getNameWithNamespaceInclAsString(),
+
pruneTable.getNameWithNamespaceInclAsString()));
}
}
}
}
+ protected void resetPruneState() {
+pruneEnable = false;
+if (compactionState != null) {
+ compactionState.stop();
--- End diff --
It would be good to reset `compactionState` to `null` here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100932301
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java
---
@@ -356,8 +356,8 @@ public InternalScanner
preCompactScannerOpen(ObserverContext e,
Store store, StoreFile resultFile,
CompactionRequest request) throws IOException {
-// Persist the compaction state after a succesful compaction
-if (compactionState != null) {
+// Persist the compaction state after a successful compaction
+if (Boolean.TRUE.equals(pruneEnable)) {
--- End diff --
It would be better to change this check to `compactionState != null`.
Otherwise there is a race condition during `initializePruneState()`, where
`pruneEnable` could be `true`, but `compactionState` is still not initialized.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/35#discussion_r100931439 --- Diff: tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java --- @@ -340,12 +345,7 @@ public InternalScanner preCompactScannerOpen(ObserverContext
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100931390
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java
---
@@ -454,28 +441,39 @@ protected Filter getTransactionFilter(Transaction tx,
ScanType type, Filter filt
return TransactionFilters.getVisibilityFilter(tx, ttlByFamily,
allowEmptyValues, type, filter);
}
- private void
initPruneState(ObserverContext c) {
-Configuration conf = getConfiguration(c.getEnvironment());
-// Configuration won't be null in TransactionProcessor but the derived
classes might return
-// null if it is not available temporarily
+ protected void initializePruneState(RegionCoprocessorEnvironment env) {
+Configuration conf = getConfiguration(env);
if (conf != null) {
pruneEnable =
conf.getBoolean(TxConstants.TransactionPruning.PRUNE_ENABLE,
TxConstants.TransactionPruning.DEFAULT_PRUNE_ENABLE);
+
if (Boolean.TRUE.equals(pruneEnable)) {
-String pruneTable =
conf.get(TxConstants.TransactionPruning.PRUNE_STATE_TABLE,
-
TxConstants.TransactionPruning.DEFAULT_PRUNE_STATE_TABLE);
-long pruneFlushInterval = TimeUnit.SECONDS.toMillis(
- conf.getLong(TxConstants.TransactionPruning.PRUNE_FLUSH_INTERVAL,
-
TxConstants.TransactionPruning.DEFAULT_PRUNE_FLUSH_INTERVAL));
-compactionState = new CompactionState(c.getEnvironment(),
TableName.valueOf(pruneTable), pruneFlushInterval);
+// pruneTable and pruneFlushInterval cannot be changed by simply
loading the Configuration dynamically
+// since we have only one flush thread across all regions and we
might loose.
--- End diff --
This comment can be removed now
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100709175
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/txprune/PruneUpperBoundWriter.java
---
@@ -81,14 +84,23 @@ private void startFlushThread() {
flushThread = new Thread("tephra-prune-upper-bound-writer") {
@Override
public void run() {
-while ((!isInterrupted()) && isRunning()) {
+Service.State serviceState = state();
+while ((!isInterrupted()) &&
+ (serviceState.equals(Service.State.NEW) ||
serviceState.equals(Service.State.STARTING) ||
--- End diff --
hmm - do you think it would simplify the code if we go back to using a
`stop` flag?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request: https://github.com/apache/incubator-tephra/pull/35#discussion_r100709097 --- Diff: tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/coprocessor/TransactionProcessor.java --- @@ -317,31 +317,31 @@ public InternalScanner preCompactScannerOpen(ObserverContext
[GitHub] incubator-tephra pull request #35: (TEPHRA-219) Execute cross region calls i...
Github user poornachandra commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/35#discussion_r100709949
--- Diff:
tephra-hbase-compat-1.1-base/src/main/java/org/apache/tephra/hbase/txprune/PruneUpperBoundWriter.java
---
@@ -81,14 +84,23 @@ private void startFlushThread() {
flushThread = new Thread("tephra-prune-upper-bound-writer") {
@Override
public void run() {
-while ((!isInterrupted()) && isRunning()) {
+Service.State serviceState = state();
+while ((!isInterrupted()) &&
+ (serviceState.equals(Service.State.NEW) ||
serviceState.equals(Service.State.STARTING) ||
+serviceState.equals(Service.State.RUNNING))) {
long now = System.currentTimeMillis();
if (now > (lastChecked + pruneFlushInterval)) {
// should flush data
try {
while (pruneEntries.firstEntry() != null) {
-Map.Entry<byte[], Long> firstEntry =
pruneEntries.firstEntry();
-
dataJanitorState.savePruneUpperBoundForRegion(firstEntry.getKey(),
firstEntry.getValue());
+final Map.Entry<byte[], Long> firstEntry =
pruneEntries.firstEntry();
+User.runAsLoginUser(new PrivilegedExceptionAction() {
--- End diff --
Adding `User.runAsLoginUser` here brings in unnecessary security code into
`PruneUpperBoundWriter` class. Also if we modify `TransactionProcessor` later
to add other table operations, we may miss to wrap those calls. I think it
would be better to do the wrapping in `TransactionProcessor` class itself. What
do you say?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra issue #34: TEPHRA-216 Handle empty transactional regions du...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/34 @gokulavasan I have ported the changes to other compat modules, please take a look. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra issue #34: TEPHRA-216 Handle empty transactional regions du...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/34 Thanks @anew. I will squash the commits and then port changes to other compat modules. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra issue #34: TEPHRA-216 Handle empty transactional regions du...
Github user poornachandra commented on the issue: https://github.com/apache/incubator-tephra/pull/34 @anew @gokulavasan In addition to addressing comments, I have added two more commits, please take a look --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
