You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2020/12/30 18:24:13 UTC
[accumulo] branch main updated (9a7749f -> 658cd00)
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git.
from 9a7749f Merge branch '1.10' into main
add 6bf3f40 Reduce log level of replication offline retries (#1843)
new 658cd00 Merge branch '1.10' into main
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/accumulo/master/replication/FinishedWorkUpdater.java | 4 ++--
.../accumulo/master/replication/RemoveCompleteReplicationRecords.java | 2 +-
.../org/apache/accumulo/master/replication/ReplicationDriver.java | 2 +-
.../main/java/org/apache/accumulo/master/replication/WorkDriver.java | 2 +-
.../main/java/org/apache/accumulo/master/replication/WorkMaker.java | 2 +-
5 files changed, 6 insertions(+), 6 deletions(-)
[accumulo] 01/01: Merge branch '1.10' into main
Posted by ct...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 658cd00f0448e48c942df21dc526049f33901f62
Merge: 9a7749f 6bf3f40
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Wed Dec 30 13:11:44 2020 -0500
Merge branch '1.10' into main
.../org/apache/accumulo/master/replication/FinishedWorkUpdater.java | 4 ++--
.../accumulo/master/replication/RemoveCompleteReplicationRecords.java | 2 +-
.../org/apache/accumulo/master/replication/ReplicationDriver.java | 2 +-
.../main/java/org/apache/accumulo/master/replication/WorkDriver.java | 2 +-
.../main/java/org/apache/accumulo/master/replication/WorkMaker.java | 2 +-
5 files changed, 6 insertions(+), 6 deletions(-)
diff --cc server/manager/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
index 57e4118,0000000..0559b31
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
+++ b/server/manager/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
@@@ -1,189 -1,0 +1,189 @@@
+/*
+ * 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.accumulo.master.replication;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.SortedMap;
+
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.protobuf.ProtobufUtil;
+import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
+import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.server.replication.StatusUtil;
+import org.apache.accumulo.server.replication.proto.Replication.Status;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+/**
+ * Update the status record in the replication table with work that has been replicated to each
+ * configured peer.
+ */
+public class FinishedWorkUpdater implements Runnable {
+ private static final Logger log = LoggerFactory.getLogger(FinishedWorkUpdater.class);
+
+ private final AccumuloClient client;
+
+ public FinishedWorkUpdater(AccumuloClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public void run() {
- log.debug("Looking for finished replication work");
++ log.trace("Looking for finished replication work");
+
+ if (!ReplicationTable.isOnline(client)) {
- log.debug("Replication table is not yet online, will retry");
++ log.trace("Replication table is not yet online, will retry");
+ return;
+ }
+
+ BatchScanner bs;
+ BatchWriter replBw;
+ try {
+ bs = ReplicationTable.getBatchScanner(client, 4);
+ replBw = ReplicationTable.getBatchWriter(client);
+ } catch (ReplicationTableOfflineException e) {
+ log.debug("Table is no longer online, will retry");
+ return;
+ }
+
+ IteratorSetting cfg = new IteratorSetting(50, WholeRowIterator.class);
+ bs.addScanIterator(cfg);
+ WorkSection.limit(bs);
+ bs.setRanges(Collections.singleton(new Range()));
+
+ try {
+ for (Entry<Key,Value> serializedRow : bs) {
+ SortedMap<Key,Value> wholeRow;
+ try {
+ wholeRow = WholeRowIterator.decodeRow(serializedRow.getKey(), serializedRow.getValue());
+ } catch (IOException e) {
+ log.warn("Could not deserialize whole row with key {}",
+ serializedRow.getKey().toStringNoTruncate(), e);
+ continue;
+ }
+
+ log.debug("Processing work progress for {} with {} columns",
+ serializedRow.getKey().getRow(), wholeRow.size());
+
+ Map<TableId,Long> tableIdToProgress = new HashMap<>();
+ boolean error = false;
+ Text buffer = new Text();
+
+ // We want to determine what the minimum point that all Work entries have replicated to
+ for (Entry<Key,Value> entry : wholeRow.entrySet()) {
+ Status status;
+ try {
+ status = Status.parseFrom(entry.getValue().get());
+ } catch (InvalidProtocolBufferException e) {
+ log.warn("Could not deserialize protobuf for {}", entry.getKey(), e);
+ error = true;
+ break;
+ }
+
+ // Get the replication target for the work record
+ entry.getKey().getColumnQualifier(buffer);
+ ReplicationTarget target = ReplicationTarget.from(buffer);
+
+ // Initialize the value in the map if we don't have one
+ if (!tableIdToProgress.containsKey(target.getSourceTableId())) {
+ tableIdToProgress.put(target.getSourceTableId(), Long.MAX_VALUE);
+ }
+
+ // Find the minimum value for begin (everyone has replicated up to this offset in the
+ // file)
+ tableIdToProgress.put(target.getSourceTableId(),
+ Math.min(tableIdToProgress.get(target.getSourceTableId()), status.getBegin()));
+ }
+
+ if (error) {
+ continue;
+ }
+
+ // Update the replication table for each source table we found work records for
+ for (Entry<TableId,Long> entry : tableIdToProgress.entrySet()) {
+ // If the progress is 0, then no one has replicated anything, and we don't need to update
+ // anything
+ if (entry.getValue() == 0) {
+ continue;
+ }
+
+ serializedRow.getKey().getRow(buffer);
+
+ log.debug("For {}, source table ID {} has replicated through {}",
+ serializedRow.getKey().getRow(), entry.getKey(), entry.getValue());
+
+ Mutation replMutation = new Mutation(buffer);
+
+ // Set that we replicated at least this much data, ignoring the other fields
+ Status updatedStatus = StatusUtil.replicated(entry.getValue());
+ Value serializedUpdatedStatus = ProtobufUtil.toValue(updatedStatus);
+
+ // Pull the sourceTableId into a Text
+ TableId srcTableId = entry.getKey();
+
+ // Make the mutation
+ StatusSection.add(replMutation, srcTableId, serializedUpdatedStatus);
+
+ log.debug("Updating replication status entry for {} with {}",
+ serializedRow.getKey().getRow(), ProtobufUtil.toString(updatedStatus));
+
+ try {
+ replBw.addMutation(replMutation);
+ } catch (MutationsRejectedException e) {
+ log.error("Error writing mutations to update replication Status"
+ + " messages in StatusSection, will retry", e);
+ return;
+ }
+ }
+ }
+ } finally {
+ log.debug("Finished updating files with completed replication work");
+
+ bs.close();
+
+ try {
+ replBw.close();
+ } catch (MutationsRejectedException e) {
+ log.error("Error writing mutations to update replication Status"
+ + " messages in StatusSection, will retry", e);
+ }
+ }
+ }
+
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
index 35418f5,0000000..1049215
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
+++ b/server/manager/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
@@@ -1,226 -1,0 +1,226 @@@
+/*
+ * 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.accumulo.master.replication;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.SortedMap;
+
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
+import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
+import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.server.replication.StatusUtil;
+import org.apache.accumulo.server.replication.proto.Replication.Status;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+/**
+ * Delete replication entries from the replication table that are fully replicated and closed
+ */
+public class RemoveCompleteReplicationRecords implements Runnable {
+ private static final Logger log = LoggerFactory.getLogger(RemoveCompleteReplicationRecords.class);
+
+ private AccumuloClient client;
+
+ public RemoveCompleteReplicationRecords(AccumuloClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public void run() {
+ BatchScanner bs;
+ BatchWriter bw;
+ try {
+ bs = ReplicationTable.getBatchScanner(client, 4);
+ bw = ReplicationTable.getBatchWriter(client);
+
+ if (bs == null || bw == null)
+ throw new AssertionError("Inconceivable; an exception should have been"
+ + " thrown, but 'bs' or 'bw' was null instead");
+ } catch (ReplicationTableOfflineException e) {
- log.debug("Not attempting to remove complete replication records as the"
++ log.trace("Not attempting to remove complete replication records as the"
+ + " table ({}) isn't yet online", ReplicationTable.NAME);
+ return;
+ }
+
+ bs.setRanges(Collections.singleton(new Range()));
+ IteratorSetting cfg = new IteratorSetting(50, WholeRowIterator.class);
+ StatusSection.limit(bs);
+ WorkSection.limit(bs);
+ bs.addScanIterator(cfg);
+
+ long recordsRemoved = 0;
+ long startTime = System.nanoTime();
+ Duration duration;
+ try {
+ recordsRemoved = removeCompleteRecords(client, bs, bw);
+ } finally {
+ bs.close();
+ try {
+ bw.close();
+ } catch (MutationsRejectedException e) {
+ log.error("Error writing mutations to {}, will retry", ReplicationTable.NAME, e);
+ }
+ duration = Duration.ofNanos(System.nanoTime() - startTime);
+ }
+
+ log.info("Removed {} complete replication entries from the table {} in {}", recordsRemoved,
+ ReplicationTable.NAME, duration);
+ }
+
+ /**
+ * Removes {@link Status} records read from the given {@code bs} and writes a delete, using the
+ * given {@code bw}, when that {@link Status} is fully replicated and closed, as defined by
+ * {@link StatusUtil#isSafeForRemoval(org.apache.accumulo.server.replication.proto.Replication.Status)}.
+ *
+ * @param client
+ * Accumulo client
+ * @param bs
+ * A BatchScanner to read replication status records from
+ * @param bw
+ * A BatchWriter to write deletes to
+ * @return Number of records removed
+ */
+ protected long removeCompleteRecords(AccumuloClient client, BatchScanner bs, BatchWriter bw) {
+ Text row = new Text(), colf = new Text(), colq = new Text();
+ long recordsRemoved = 0;
+
+ // For each row in the replication table
+ for (Entry<Key,Value> rowEntry : bs) {
+ SortedMap<Key,Value> columns;
+ try {
+ columns = WholeRowIterator.decodeRow(rowEntry.getKey(), rowEntry.getValue());
+ } catch (IOException e) {
+ log.error("Could not deserialize {} with WholeRowIterator", rowEntry.getKey().getRow(), e);
+ continue;
+ }
+
+ rowEntry.getKey().getRow(row);
+
+ // Try to remove the row (all or nothing)
+ recordsRemoved += removeRowIfNecessary(bw, columns, row, colf, colq);
+ }
+
+ return recordsRemoved;
+ }
+
+ protected long removeRowIfNecessary(BatchWriter bw, SortedMap<Key,Value> columns, Text row,
+ Text colf, Text colq) {
+ long recordsRemoved = 0;
+ if (columns.isEmpty()) {
+ return recordsRemoved;
+ }
+
+ Mutation m = new Mutation(row);
+ Map<TableId,Long> tableToTimeCreated = new HashMap<>();
+ for (Entry<Key,Value> entry : columns.entrySet()) {
+ Status status = null;
+ try {
+ status = Status.parseFrom(entry.getValue().get());
+ } catch (InvalidProtocolBufferException e) {
+ log.error("Encountered unparsable protobuf for key: {}",
+ entry.getKey().toStringNoTruncate());
+ continue;
+ }
+
+ // If a column in the row isn't ready for removal, we keep the whole row
+ if (!StatusUtil.isSafeForRemoval(status)) {
+ return 0L;
+ }
+
+ Key k = entry.getKey();
+ k.getColumnFamily(colf);
+ k.getColumnQualifier(colq);
+
+ log.debug("Removing {} {}:{} from replication table", row, colf, colq);
+
+ m.putDelete(colf, colq);
+
+ TableId tableId;
+ if (StatusSection.NAME.equals(colf)) {
+ tableId = TableId.of(colq.toString());
+ } else if (WorkSection.NAME.equals(colf)) {
+ ReplicationTarget target = ReplicationTarget.from(colq);
+ tableId = target.getSourceTableId();
+ } else {
+ throw new RuntimeException("Got unexpected column");
+ }
+
+ if (status.hasCreatedTime()) {
+ Long timeClosed = tableToTimeCreated.get(tableId);
+ if (timeClosed == null) {
+ tableToTimeCreated.put(tableId, status.getCreatedTime());
+ } else if (timeClosed != status.getCreatedTime()) {
+ log.warn("Found multiple values for timeClosed for {}: {} and {}", row, timeClosed,
+ status.getCreatedTime());
+ }
+ }
+
+ recordsRemoved++;
+ }
+
+ List<Mutation> mutations = new ArrayList<>();
+ mutations.add(m);
+ for (Entry<TableId,Long> entry : tableToTimeCreated.entrySet()) {
+ log.info("Removing order mutation for table {} at {} for {}", entry.getKey(),
+ entry.getValue(), row);
+ Mutation orderMutation = OrderSection.createMutation(row.toString(), entry.getValue());
+ orderMutation.putDelete(OrderSection.NAME, new Text(entry.getKey().canonical()));
+ mutations.add(orderMutation);
+ }
+
+ // Send the mutation deleting all the columns at once.
+ // If we send them not as a single Mutation, we run the risk of having some of them be applied
+ // which would mean that we might accidentally re-replicate data. We want to get rid of them all
+ // at once
+ // or not at all.
+ try {
+ bw.addMutations(mutations);
+ bw.flush();
+ } catch (MutationsRejectedException e) {
+ log.error("Could not submit mutation to remove columns for {} in replication table", row, e);
+ return 0L;
+ }
+
+ return recordsRemoved;
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/master/replication/ReplicationDriver.java
index 8050838,0000000..5d1e1fc
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/master/replication/ReplicationDriver.java
+++ b/server/manager/src/main/java/org/apache/accumulo/master/replication/ReplicationDriver.java
@@@ -1,124 -1,0 +1,124 @@@
+/*
+ * 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.accumulo.master.replication;
+
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.core.util.Daemon;
+import org.apache.accumulo.fate.util.UtilWaitThread;
+import org.apache.accumulo.master.Master;
+import org.apache.htrace.Trace;
+import org.apache.htrace.TraceScope;
+import org.apache.htrace.impl.ProbabilitySampler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Daemon wrapper around the {@link WorkMaker} that separates it from the Master
+ */
+public class ReplicationDriver extends Daemon {
+ private static final Logger log = LoggerFactory.getLogger(ReplicationDriver.class);
+
+ private final Master master;
+ private final AccumuloConfiguration conf;
+
+ private WorkMaker workMaker;
+ private StatusMaker statusMaker;
+ private FinishedWorkUpdater finishedWorkUpdater;
+ private RemoveCompleteReplicationRecords rcrr;
+ private AccumuloClient client;
+
+ public ReplicationDriver(Master master) {
+ super("Replication Driver");
+
+ this.master = master;
+ this.conf = master.getConfiguration();
+ }
+
+ @Override
+ public void run() {
+ ProbabilitySampler sampler =
+ TraceUtil.probabilitySampler(conf.getFraction(Property.REPLICATION_TRACE_PERCENT));
+
+ long millisToWait = conf.getTimeInMillis(Property.REPLICATION_DRIVER_DELAY);
+ log.debug("Waiting {}ms before starting main replication loop", millisToWait);
+ UtilWaitThread.sleep(millisToWait);
+
+ log.debug("Starting replication loop");
+
+ while (master.stillMaster()) {
+ if (workMaker == null) {
+ client = master.getContext();
+ statusMaker = new StatusMaker(client, master.getVolumeManager());
+ workMaker = new WorkMaker(master.getContext(), client);
+ finishedWorkUpdater = new FinishedWorkUpdater(client);
+ rcrr = new RemoveCompleteReplicationRecords(client);
+ }
+
+ try (TraceScope replicationDriver = Trace.startSpan("masterReplicationDriver", sampler)) {
+
+ // Make status markers from replication records in metadata, removing entries in
+ // metadata which are no longer needed (closed records)
+ // This will end up creating the replication table too
+ try {
+ statusMaker.run();
+ } catch (Exception e) {
+ log.error("Caught Exception trying to create Replication status records", e);
+ }
+
+ // Tell the work maker to make work
+ try {
+ workMaker.run();
+ } catch (Exception e) {
+ log.error("Caught Exception trying to create Replication work records", e);
+ }
+
+ // Update the status records from the work records
+ try {
+ finishedWorkUpdater.run();
+ } catch (Exception e) {
+ log.error(
+ "Caught Exception trying to update Replication records using finished work records",
+ e);
+ }
+
+ // Clean up records we no longer need.
+ // It must be running at the same time as the StatusMaker or WorkMaker
+ // So it's important that we run these sequentially and not concurrently
+ try {
+ rcrr.run();
+ } catch (Exception e) {
+ log.error("Caught Exception trying to remove finished Replication records", e);
+ }
+
+ }
+
+ // Sleep for a bit
+ long sleepMillis = conf.getTimeInMillis(Property.MASTER_REPLICATION_SCAN_INTERVAL);
- log.debug("Sleeping for {}ms before re-running", sleepMillis);
++ log.trace("Sleeping for {}ms before re-running", sleepMillis);
+ try {
+ Thread.sleep(sleepMillis);
+ } catch (InterruptedException e) {
+ log.error("Interrupted while sleeping", e);
+ }
+ }
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
index 928b920,0000000..02e6b94
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
+++ b/server/manager/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
@@@ -1,96 -1,0 +1,96 @@@
+/*
+ * 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.accumulo.master.replication;
+
+import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.util.Daemon;
+import org.apache.accumulo.master.Master;
+import org.apache.accumulo.server.replication.WorkAssigner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Driver for a {@link WorkAssigner}
+ */
+public class WorkDriver extends Daemon {
+ private static final Logger log = LoggerFactory.getLogger(WorkDriver.class);
+
+ private Master master;
+ private AccumuloClient client;
+ private AccumuloConfiguration conf;
+
+ private WorkAssigner assigner;
+ private String assignerImplName;
+
+ public WorkDriver(Master master) {
+ super();
+ this.master = master;
+ this.client = master.getContext();
+ this.conf = master.getConfiguration();
+ configureWorkAssigner();
+ }
+
+ protected void configureWorkAssigner() {
+ String workAssignerClass = conf.get(Property.REPLICATION_WORK_ASSIGNER);
+
+ if (assigner == null || !assigner.getClass().getName().equals(workAssignerClass)) {
+ log.info("Initializing work assigner implementation of {}", workAssignerClass);
+
+ try {
+ Class<?> clz = Class.forName(workAssignerClass);
+ Class<? extends WorkAssigner> workAssignerClz = clz.asSubclass(WorkAssigner.class);
+ this.assigner = workAssignerClz.getDeclaredConstructor().newInstance();
+ } catch (ReflectiveOperationException e) {
+ log.error("Could not instantiate configured work assigner {}", workAssignerClass, e);
+ throw new RuntimeException(e);
+ }
+
+ this.assigner.configure(conf, client);
+ this.assignerImplName = assigner.getClass().getName();
+ this.setName(assigner.getName());
+ }
+ }
+
+ @Override
+ public void run() {
+ log.info("Starting replication work assignment thread using {}", assignerImplName);
+
+ while (master.stillMaster()) {
+ // Assign the work using the configured implementation
+ try {
+ assigner.assignWork();
+ } catch (Exception e) {
+ log.error("Error while assigning work", e);
+ }
+
+ long sleepTime = conf.getTimeInMillis(Property.REPLICATION_WORK_ASSIGNMENT_SLEEP);
- log.debug("Sleeping {} ms before next work assignment", sleepTime);
++ log.trace("Sleeping {} ms before next work assignment", sleepTime);
+ sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
+
+ // After each loop, make sure that the WorkAssigner implementation didn't change
+ configureWorkAssigner();
+ }
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
index a222196,0000000..81ea349
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
+++ b/server/manager/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
@@@ -1,209 -1,0 +1,209 @@@
+/*
+ * 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.accumulo.master.replication;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.replication.ReplicationSchema;
+import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
+import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.conf.TableConfiguration;
+import org.apache.accumulo.server.replication.StatusUtil;
+import org.apache.accumulo.server.replication.proto.Replication.Status;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.Text;
+import org.apache.htrace.Trace;
+import org.apache.htrace.TraceScope;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+/**
+ * Reads replication records from the replication table and creates work records which include
+ * target replication system information.
+ */
+public class WorkMaker {
+ private static final Logger log = LoggerFactory.getLogger(WorkMaker.class);
+
+ private final ServerContext context;
+ private AccumuloClient client;
+
+ private BatchWriter writer;
+
+ public WorkMaker(ServerContext context, AccumuloClient client) {
+ this.context = context;
+ this.client = client;
+ }
+
+ public void run() {
+ if (!ReplicationTable.isOnline(client)) {
- log.debug("Replication table is not yet online");
++ log.trace("Replication table is not yet online");
+ return;
+ }
+
+ try (TraceScope span = Trace.startSpan("replicationWorkMaker")) {
+ final Scanner s;
+ try {
+ s = ReplicationTable.getScanner(client);
+ if (writer == null) {
+ setBatchWriter(ReplicationTable.getBatchWriter(client));
+ }
+ } catch (ReplicationTableOfflineException e) {
+ log.warn("Replication table was online, but not anymore");
+ writer = null;
+ return;
+ }
+
+ // Only pull records about data that has been ingested and is ready for replication
+ StatusSection.limit(s);
+
+ TableConfiguration tableConf;
+
+ Text file = new Text();
+ for (Entry<Key,Value> entry : s) {
+ // Extract the useful bits from the status key
+ ReplicationSchema.StatusSection.getFile(entry.getKey(), file);
+ TableId tableId = ReplicationSchema.StatusSection.getTableId(entry.getKey());
+ log.debug("Processing replication status record for {} on table {}", file, tableId);
+
+ Status status;
+ try {
+ status = Status.parseFrom(entry.getValue().get());
+ } catch (InvalidProtocolBufferException e) {
+ log.error("Could not parse protobuf for {} from table {}", file, tableId);
+ continue;
+ }
+
+ // Don't create the record if we have nothing to do.
+ // TODO put this into a filter on serverside
+ if (!shouldCreateWork(status)) {
+ log.debug("Not creating work: {}", status);
+ continue;
+ }
+
+ // Get the table configuration for the table specified by the status record
+ tableConf = context.getTableConfiguration(tableId);
+
+ // getTableConfiguration(String) returns null if the table no longer exists
+ if (tableConf == null) {
+ continue;
+ }
+
+ // Pull the relevant replication targets
+ // TODO Cache this instead of pulling it every time
+ Map<String,String> replicationTargets = getReplicationTargets(tableConf);
+
+ // If we have targets, we need to make a work record
+ // TODO Don't replicate if it's a only a newFile entry (nothing to replicate yet)
+ // -- Another scanner over the WorkSection can make this relatively cheap
+ if (replicationTargets.isEmpty()) {
+ log.warn("No configured targets for table with ID {}", tableId);
+ } else {
+ try (TraceScope workSpan = Trace.startSpan("createWorkMutations")) {
+ addWorkRecord(file, entry.getValue(), replicationTargets, tableId);
+ }
+ }
+ }
+ }
+ }
+
+ protected void setBatchWriter(BatchWriter bw) {
+ this.writer = bw;
+ }
+
+ protected Map<String,String> getReplicationTargets(TableConfiguration tableConf) {
+ final Map<String,String> props =
+ tableConf.getAllPropertiesWithPrefix(Property.TABLE_REPLICATION_TARGET);
+ final Map<String,String> targets = new HashMap<>();
+ final int propKeyLength = Property.TABLE_REPLICATION_TARGET.getKey().length();
+
+ for (Entry<String,String> prop : props.entrySet()) {
+ targets.put(prop.getKey().substring(propKeyLength), prop.getValue());
+ }
+
+ return targets;
+ }
+
+ /**
+ * @return Should a Work entry be created for this status
+ */
+ protected boolean shouldCreateWork(Status status) {
+ // Only creating work when there is work to do (regardless of closed status) is safe
+ // as long as the ReplicaSystem implementation is correctly observing
+ // that a file is completely replicated only when the file is closed
+ return StatusUtil.isWorkRequired(status);
+ }
+
+ protected void addWorkRecord(Text file, Value v, Map<String,String> targets,
+ TableId sourceTableId) {
+ log.info("Adding work records for {} to targets {}", file, targets);
+ try {
+ Mutation m = new Mutation(file);
+
+ ReplicationTarget target = new ReplicationTarget();
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ Text t = new Text();
+ for (Entry<String,String> entry : targets.entrySet()) {
+ buffer.reset();
+
+ // Set up the writable
+ target.setPeerName(entry.getKey());
+ target.setRemoteIdentifier(entry.getValue());
+ target.setSourceTableId(sourceTableId);
+ target.write(buffer);
+
+ // Throw it in a text for the mutation
+ t.set(buffer.getData(), 0, buffer.getLength());
+
+ // Add it to the work section
+ WorkSection.add(m, t, v);
+ }
+ try {
+ writer.addMutation(m);
+ } catch (MutationsRejectedException e) {
+ log.warn("Failed to write work mutations for replication, will retry", e);
+ }
+ } catch (IOException e) {
+ log.warn("Failed to serialize data to Text, will retry", e);
+ } finally {
+ try {
+ writer.flush();
+ } catch (MutationsRejectedException e) {
+ log.warn("Failed to write work mutations for replication, will retry", e);
+ }
+ }
+ }
+}