Github user ramkrish86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/2332#discussion_r74550914
--- Diff:
flink-streaming-connectors/flink-connector-hbase/src/main/java/org/apache/flink/streaming/connectors/hbase/MutationActionList.java
---
@@ -0,0 +1,371 @@
+/*
+ * 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.flink.streaming.connectors.hbase;
+
+import org.apache.flink.util.Preconditions;
+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.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This class represents a list of {@link MutationAction}s you will take
when writing
+ * an input value of {@link HBaseSink} to a row in a HBase table.
+ * Each {@link MutationAction} can create an HBase {@link Mutation}
operation type
+ * including {@link Put}, {@link Increment}, {@link Append} and {@link
Delete}.
+ */
+public class MutationActionList {
+ private final List<MutationAction> actions;
+
+ public MutationActionList() {
+ this.actions = new ArrayList<>();
+ }
+
+ public List<MutationAction> getActions() {
+ return this.actions;
+ }
+
+ /**
+ * Create a new list of HBase {@link Mutation}s.
+ *
+ * @param rowKey row that the created {@link Mutation} list is applied
to
+ * @param writeToWAL enable WAL
+ * @return a list of HBase {@link Mutation}s
+ */
+ public List<Mutation> newMutationList(byte[] rowKey, boolean
writeToWAL) {
+ List<Mutation> mutations = new ArrayList<>();
+ Put put = null;
+ Increment increment = null;
+ Append append = null;
+ Delete delete = null;
+ boolean rowIsDeleted = false;
+ for (MutationAction action : actions) {
+ switch (action.getType()) {
+ case PUT:
+ if (put == null) {
+ put = new Put(rowKey);
+ mutations.add(put);
+ }
+ if (action.getTs() == -1) {
+
put.addColumn(action.getFamily(), action.getQualifier(), action.getValue());
+ } else {
+
put.addColumn(action.getFamily(), action.getQualifier(), action.getTs(),
action.getValue());
+ }
+ break;
+
+ case INCREMENT:
+ if (increment == null) {
+ increment = new
Increment(rowKey);
+ mutations.add(increment);
+ }
+ increment.addColumn(action.getFamily(),
action.getQualifier(), action.getIncrement());
+ break;
+
+ case APPEND:
+ if (append == null) {
+ append = new Append(rowKey);
+ mutations.add(append);
+ }
+ append.add(action.getFamily(),
action.getQualifier(), action.getValue());
+ break;
+
+ // If there are multiple DELETE_ROW actions,
only the first one is served
+ case DELETE_ROW:
+ if (!rowIsDeleted) {
+ for (int i = 0; i <
mutations.size(); ) {
+ if (mutations.get(i)
instanceof Delete) {
+
mutations.remove(i);
+ } else {
+ i++;
+ }
+ }
+ delete = new Delete(rowKey,
action.getTs());
+ mutations.add(delete);
+ rowIsDeleted = true;
+ }
+ break;
+
+ case DELETE_FAMILY:
+ if (!rowIsDeleted) {
+ if (delete == null) {
+ delete = new
Delete(rowKey);
+ mutations.add(delete);
+ }
+
delete.addFamily(action.getFamily(), action.getTs());
+ }
+ break;
+
+ case DELETE_COLUMNS:
+ if (!rowIsDeleted) {
+ if (delete == null) {
+ delete = new
Delete(rowKey);
+ mutations.add(delete);
+ }
+
delete.addColumns(action.getFamily(), action.getQualifier(), action.getTs());
+ }
+ break;
+
+ case DELETE_COLUMN:
+ if (!rowIsDeleted) {
+ if (delete == null) {
+ delete = new
Delete(rowKey);
+ mutations.add(delete);
+ }
+
delete.addColumn(action.getFamily(), action.getQualifier(), action.getTs());
+ }
+ break;
+
+ default:
+ throw new
IllegalArgumentException("Cannot process such action type: " +
action.getType());
+ }
+ }
+ Durability durability = writeToWAL ? Durability.SYNC_WAL :
Durability.SKIP_WAL;
+ for (Mutation mutation : mutations) {
+ mutation.setDurability(durability);
+ }
+ return mutations;
+ }
+
+ /**
+ * Create a new list of HBase {@link Mutation}s enabling WAL as default.
+ *
+ * @param rowKey row that the created {@link Mutation} list is applied
to
+ * @return a list of HBase {@link Mutation}s
+ */
+ public List<Mutation> newMutationList(byte[] rowKey) {
+ return newMutationList(rowKey, true);
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#PUT}. which will
+ * create an HBase {@link Put} operation for a specified row with
specified timestamp.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @param value column value
+ * @param timestamp version timestamp
+ * @return this
+ */
+ public MutationActionList addPut(byte[] family, byte[] qualifier,
byte[] value, long timestamp) {
+ actions.add(new MutationAction(family, qualifier, value,
timestamp, 0, MutationAction.Type.PUT));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#PUT}, which will
+ * create an HBase {@link Put} operation for a specified row with no
timestamp specified at client side.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @param value column value
+ * @return this
+ */
+ public MutationActionList addPut(byte[] family, byte[] qualifier,
byte[] value) {
+ actions.add(new MutationAction(family, qualifier, value, 0,
MutationAction.Type.PUT));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#APPEND}, which will
+ * create an HBase {@link Append} operation for a specified row.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @param value column value
+ * @return this
+ */
+ public MutationActionList addAppend(byte[] family, byte[] qualifier,
byte[] value) {
+ actions.add(new MutationAction(family, qualifier, value, 0,
MutationAction.Type.APPEND));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#INCREMENT}, which will
+ * create an HBase {@link Increment} operation for a specified row with
the specified amount.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @param amount amount to increment by
+ * @return this
+ */
+ public MutationActionList addIncrement(byte[] family, byte[] qualifier,
long amount) {
+ actions.add(new MutationAction(family, qualifier, null, amount,
MutationAction.Type.INCREMENT));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_ROW}, which will
+ * create an HBase {@link Delete} operation that deletes all columns in
all families of the specified row with
+ * a timestamp less than or equal to the specified timestamp.
+ *
+ * @param timestamp version timestamp
+ * @return this
+ */
+ public MutationActionList addDeleteRow(long timestamp) {
+ actions.add(new MutationAction(null, null, null, timestamp, 0,
MutationAction.Type.DELETE_ROW));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_ROW}, which will
+ * create an HBase {@link Delete} operation that deletes everything
associated with the specified row (all versions
+ * of all columns in all families).
+ *
+ * @return this
+ */
+ public MutationActionList addDeleteRow() {
+ actions.add(new MutationAction(null, null, null,
Long.MAX_VALUE, 0, MutationAction.Type.DELETE_ROW));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_FAMILY},
+ * which will create an HBase {@link Delete} operation that deletes
versions of the specified family with
+ * timestamps less than or equal to the specified timestamp.
+ *
+ * @param family family name
+ * @param timestamp version timestamp
+ * @return this
+ */
+ public MutationActionList addDeleteFamily(byte[] family, long
timestamp) {
+ actions.add(new MutationAction(family, null, null, timestamp,
0, MutationAction.Type.DELETE_FAMILY));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_FAMILY},
+ * which will create an HBase {@link Delete} operation that deletes all
versions of the specified family.
+ *
+ * @param family family name
+ * @return this
+ */
+ public MutationActionList addDeleteFamily(byte[] family) {
+ actions.add(new MutationAction(family, null, null,
Long.MAX_VALUE, 0, MutationAction.Type.DELETE_FAMILY));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_COLUMNS},
+ * which will create an HBase {@link Delete} operation that deletes
versions of the specified column with
+ * timestamps less than or equal to the specified timestamp.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @param timestamp version timestamp
+ * @return this
+ */
+ public MutationActionList addDeleteColumns(byte[] family, byte[]
qualifier, long timestamp) {
+ actions.add(new MutationAction(family, qualifier, null,
timestamp, 0, MutationAction.Type.DELETE_COLUMNS));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_COLUMNS},
+ * which will create an HBase {@link Delete} operation that deletes all
versions of the specified column.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @return this
+ */
+ public MutationActionList addDeleteColumns(byte[] family, byte[]
qualifier) {
+ actions.add(new MutationAction(family, qualifier, null,
Long.MAX_VALUE, 0, MutationAction.Type.DELETE_COLUMNS));
+ return this;
+ }
+
+ /**
+ * Add to {@link MutationActionList} a {@link MutationAction} of type
{@link MutationAction.Type#DELETE_COLUMN},
+ * which will create an HBase {@link Delete} operation that deletes the
specified version of the specified column.
+ *
+ * @param family family name
+ * @param qualifier column qualifier
+ * @param timestamp version timestamp
+ * @return this
+ */
+ public MutationActionList addDeleteColumn(byte[] family, byte[]
qualifier, long timestamp) {
+ actions.add(new MutationAction(family, qualifier, null,
timestamp, 0, MutationAction.Type.DELETE_COLUMN));
+ return this;
+ }
+
+ static class MutationAction {
--- End diff --
Will it not be better to expose MutationAction - and allow the user to
create a MutationAction and then just add them to the MutationList.
MutationAction can have all the different flavours of addXXX() APIs? The
return type of each will be a MutationAction. That will get added to a
MutationActionList and so the getActions() in mutation Action will just return
the actions added to it. What do you think? Related to the HBaseSink
implementing RichSinkFunction I think people from Flink community can answer
you the best.
---
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.
---
