package com.cloudera.flume.agent;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.core.Attributes;
import com.cloudera.flume.handlers.endtoend.AckListener;
import com.cloudera.flume.reporter.ReportEvent;
import com.cloudera.flume.reporter.Reportable;
import com.cloudera.util.Clock;
import com.google.common.base.Preconditions;
private static final String A_RETRANSMIT_TIMEOUT = "retransmitTimeout";
private static final String A_PENDING_ACK_INFO = "pendingAckInfo";
static final Logger LOG = LoggerFactory.getLogger(WALAckManager.class);
final ConcurrentHashMap<String, Long> pending = new ConcurrentHashMap<String, Long>();
MasterRPC client;
final AckListener queuer = new PendingAckQueuer();
final AckListener listener;
final long retransmitTime;
WALAckManager(MasterRPC c, AckListener listener,
long ackRetransmit) {
Preconditions.checkNotNull(c);
Preconditions.checkNotNull(listener);
this.client = c;
this.listener = listener;
this.retransmitTime = ackRetransmit;
}
@Override
public void end(String group)
throws IOException {
long now = Clock.unixTime();
LOG.info("Ack for " + group + " is queued to be checked");
synchronized (pending) {
pending.put(group, now);
}
}
};
return queuer;
}
LOG.debug("agent acks waiting for master: " + pending);
List<String> done = new ArrayList<String>();
for (String k : pending.keySet()) {
try {
boolean acked = client.checkAck(k);
if (acked) {
done.add(k);
}
} catch (IOException e) {
LOG.error("Master connection exception", e);
}
}
for (String k : done) {
try {
listener.end(k);
pending.remove(k);
LOG.debug("removed ack tag from agent's ack queue: " + k);
} catch (IOException e) {
LOG.error("problem notifying agent pending ack queue", e);
}
}
}
long now = Clock.unixTime();
List<String> retried = new ArrayList<String>();
for (Entry<String, Long> ack : pending.entrySet()) {
if (now - ack.getValue() > retransmitTime) {
try {
LOG.info("Retransmitting " + ack.getKey());
listener.expired(ack.getKey());
retried.add(ack.getKey());
} catch (IOException e) {
LOG.error("problem notifying agent pending ack queue", e);
}
}
}
for (String key : retried) {
pending.put(key, now);
}
}
long now = Clock.unixTime();
List<String> retried = new ArrayList<String>();
for (Entry<String, Long> ack : pending.entrySet()) {
try {
LOG.info("Retransmitting " + ack.getKey());
listener.expired(ack.getKey());
retried.add(ack.getKey());
} catch (IOException e) {
LOG.error("problem notifying agent pending ack queue", e);
}
}
for (String key : retried) {
pending.put(key, now);
}
}
@Override
return "AgentWALAckManager";
}
@Override
synchronized public ReportEvent
getReport() {
ReportEvent rpt = new ReportEvent(getName());
Attributes.setLong(rpt, A_RETRANSMIT_TIMEOUT, retransmitTime);
StringBuilder pendingAcks = new StringBuilder();
for (Map.Entry<String, Long> e : pending.entrySet()) {
pendingAcks.append(e.getKey());
pendingAcks.append(":");
pendingAcks.append(new Date(e.getValue()).toString());
pendingAcks.append(", ");
}
Attributes.setString(rpt, A_PENDING_ACK_INFO, pendingAcks.toString());
return rpt;
}
return Collections.unmodifiableSet(pending.keySet());
}
}