Improve SegmentedJournalActor logging

[controller.git] / opendaylight / md-sal / sal-akka-segmented-journal / src / main / java / org / opendaylight / controller / akka / segjournal / DataJournalV0.java

diff --git a/opendaylight/md-sal/sal-akka-segmented-journal/src/main/java/org/opendaylight/controller/akka/segjournal/DataJournalV0.java b/opendaylight/md-sal/sal-akka-segmented-journal/src/main/java/org/opendaylight/controller/akka/segjournal/DataJournalV0.java
index 766f2fac05e8e46dd08f76ecf80043aa51b5f6e0..3203d6617110674d08c86e79edc72b6b6db48393 100644 (file)
--- a/opendaylight/md-sal/sal-akka-segmented-journal/src/main/java/org/opendaylight/controller/akka/segjournal/DataJournalV0.java
+++ b/opendaylight/md-sal/sal-akka-segmented-journal/src/main/java/org/opendaylight/controller/akka/segjournal/DataJournalV0.java
@@ -21,6 +21,7 @@ import io.atomix.storage.journal.SegmentedJournalWriter;
 import io.atomix.utils.serializer.Namespace;
 import java.io.File;
 import java.io.Serializable;
+import java.util.List;
 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.FromPersistence;
 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.ToPersistence;
 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.ReplayMessages;
@@ -52,18 +53,18 @@ final class DataJournalV0 extends DataJournal {
     }
 
     @Override
-    long lastWrittenIndex() {
+    long lastWrittenSequenceNr() {
         return entries.writer().getLastIndex();
     }
 
     @Override
-    void commitTo(final long index) {
-        entries.writer().commit(index);
+    void deleteTo(final long sequenceNr) {
+        entries.writer().commit(sequenceNr);
     }
 
     @Override
-    void compactTo(final long index) {
-        entries.compact(index);
+    void compactTo(final long sequenceNr) {
+        entries.compact(sequenceNr + 1);
     }
 
     @Override
@@ -73,8 +74,8 @@ final class DataJournalV0 extends DataJournal {
 
     @Override
     @SuppressWarnings("checkstyle:illegalCatch")
-    void handleReplayMessages(final ReplayMessages message, final long from) {
-        try (SegmentedJournalReader<DataJournalEntry> reader = entries.openReader(from)) {
+    void handleReplayMessages(final ReplayMessages message, final long fromSequenceNr) {
+        try (SegmentedJournalReader<DataJournalEntry> reader = entries.openReader(fromSequenceNr)) {
             int count = 0;
             while (reader.hasNext() && count < message.max) {
                 final Indexed<DataJournalEntry> next = reader.next();
@@ -103,25 +104,21 @@ final class DataJournalV0 extends DataJournal {
 
     @Override
     @SuppressWarnings("checkstyle:illegalCatch")
-    void handleWriteMessages(final WriteMessages message) {
+    long handleWriteMessages(final WriteMessages message) {
         final int count = message.size();
         final SegmentedJournalWriter<DataJournalEntry> writer = entries.writer();
+        long bytes = 0;
 
         for (int i = 0; i < count; ++i) {
             final long mark = writer.getLastIndex();
             final AtomicWrite request = message.getRequest(i);
+
+            final List<PersistentRepr> reprs = CollectionConverters.asJava(request.payload());
+            LOG.trace("{}: append {}/{}: {} items at mark {}", persistenceId, i, count, reprs.size(), mark);
             try {
-                for (PersistentRepr repr : CollectionConverters.asJava(request.payload())) {
-                    final Object payload = repr.payload();
-                    if (!(payload instanceof Serializable)) {
-                        throw new UnsupportedOperationException("Non-serializable payload encountered "
-                                + payload.getClass());
-                    }
-
-                    recordMessageSize(writer.append(new ToPersistence(repr)).size());
-                }
+                bytes += writePayload(writer, reprs);
             } catch (Exception e) {
-                LOG.warn("{}: failed to write out request", persistenceId, e);
+                LOG.warn("{}: failed to write out request {}/{} reverting to {}", persistenceId, i, count, mark, e);
                 message.setFailure(i, e);
                 writer.truncate(mark);
                 continue;
@@ -130,5 +127,25 @@ final class DataJournalV0 extends DataJournal {
             message.setSuccess(i);
         }
         writer.flush();
+        return bytes;
+    }
+
+    private long writePayload(final SegmentedJournalWriter<DataJournalEntry> writer, final List<PersistentRepr> reprs) {
+        long bytes = 0;
+        for (PersistentRepr repr : reprs) {
+            final Object payload = repr.payload();
+            if (!(payload instanceof Serializable)) {
+                throw new UnsupportedOperationException("Non-serializable payload encountered "
+                        + payload.getClass());
+            }
+
+            LOG.trace("{}: starting append of {}", persistenceId, payload);
+            final Indexed<ToPersistence> entry = writer.append(new ToPersistence(repr));
+            final int size = entry.size();
+            LOG.trace("{}: finished append of {} with {} bytes at {}", persistenceId, payload, size, entry.index());
+            recordMessageSize(size);
+            bytes += size;
+        }
+        return bytes;
     }
 }