Added Message.setRollbackOnly() method to rollback message

killme2008 · Apr 3, 2013 · 0f8c7fc · 0f8c7fc
1 parent 0a617eb
commit 0f8c7fc
Show file tree

Hide file tree

Showing 4 changed files with 74 additions and 1 deletion.
diff --git a/...sis-client/src/main/java/com/taobao/metamorphosis/client/consumer/SimpleFetchManager.java b/...sis-client/src/main/java/com/taobao/metamorphosis/client/consumer/SimpleFetchManager.java
@@ -353,6 +353,11 @@ private boolean processReceiveMessage(final FetchRequest request, final MessageI
                     final Message msg = it.next();
                     MessageAccessor.setPartition(msg, partition);
                     listener.recieveMessages(msg);
+                    // rollback message if it is in rollback only state.
+                    if (MessageAccessor.isRollbackOnly(msg)) {
+                        it.setOffset(prevOffset);
+                        break;
+                    }
                     if (partition.isAutoAck()) {
                         count++;
                     }

diff --git a/...nt/src/test/java/com/taobao/metamorphosis/client/consumer/SimpleFetchManagerUnitTest.java b/...nt/src/test/java/com/taobao/metamorphosis/client/consumer/SimpleFetchManagerUnitTest.java
@@ -296,6 +296,50 @@ public Executor getExecutor() {
     }
 
 
+    @Test
+    public void testProcessRequestMessageRollbackOnly() throws Exception {
+        // this.mockConsumerReInitializeFetchManager();
+        final String topic = "topic1";
+        final int maxSize = 1024;
+        final Partition partition = new Partition("0-0");
+        final long offset = 12;
+        final Broker broker = new Broker(0, "meta://localhost:0");
+        final byte[] data =
+                MessageUtils.makeMessageBuffer(1111,
+                    new PutCommand(topic, partition.getPartition(), "hello".getBytes(), null, 0, 0)).array();
+        final FetchRequest request =
+                new FetchRequest(broker, 0, new TopicPartitionRegInfo(topic, partition, offset), maxSize);
+
+        final FetchRequestRunner runner = this.fetchManager.new FetchRequestRunner();
+
+        EasyMock.expect(this.consumer.fetch(request, -1, null)).andReturn(new MessageIterator(topic, data));
+        EasyMock.expect(this.consumer.getMessageListener(topic)).andReturn(new MessageListener() {
+
+            @Override
+            public void recieveMessages(final Message message) {
+                System.out.println("Rollback current message");
+                message.setRollbackOnly();
+            }
+
+
+            @Override
+            public Executor getExecutor() {
+                return null;
+            }
+        });
+
+        final FetchRequest newRequest =
+                new FetchRequest(broker, this.consumerConfig.getMaxDelayFetchTimeInMills() / 10,
+                    new TopicPartitionRegInfo(topic, partition, offset, 1111), maxSize);
+        newRequest.incrementRetriesAndGet();
+
+        EasyMock.replay(this.consumer);
+        runner.processRequest(request);
+        EasyMock.verify(this.consumer);
+        assertEquals(newRequest, this.fetchManager.takeFetchRequest());
+    }
+
+
     @Test
     public void testProcessRequestDelayed_IncreaseMaxSize() throws Exception {
         // this.mockConsumerReInitializeFetchManager();

diff --git a/metamorphosis-commons/src/main/java/com/taobao/metamorphosis/Message.java b/metamorphosis-commons/src/main/java/com/taobao/metamorphosis/Message.java
@@ -43,6 +43,8 @@ public class Message implements Serializable {
     // added by dennis<killme2008@gmail.com>,2012-06-14
     private transient boolean readOnly;
 
+    private transient boolean rollbackOnly = false;
+
 
     void setId(final long id) {
         this.id = id;
@@ -66,6 +68,22 @@ void setFlag(final int flag) {
     }
 
 
+    boolean isRollbackOnly() {
+        return this.rollbackOnly;
+    }
+
+
+    /**
+     * Set message to be in rollback only state.The state is transient,it's only
+     * valid in current message instance.
+     * 
+     * @since 1.4.5
+     */
+    public void setRollbackOnly() {
+        this.rollbackOnly = true;
+    }
+
+
     /**
      * Returns whether the message is readonly.
      * 
@@ -79,7 +97,8 @@ public boolean isReadOnly() {
 
     /**
      * Set the message to be readonly,but metamorphosis client and server could
-     * modify message's id,flag,partition.
+     * modify message's id,flag,partition.The readonly state is transient,it
+     * will not be persist in broker.
      * 
      * @since 1.4.4
      * @param readOnly

diff --git a/metamorphosis-commons/src/main/java/com/taobao/metamorphosis/MessageAccessor.java b/metamorphosis-commons/src/main/java/com/taobao/metamorphosis/MessageAccessor.java
@@ -38,6 +38,11 @@ public static void setFlag(Message message, int flag) {
     }
 
 
+    public static boolean isRollbackOnly(Message message) {
+        return message.isRollbackOnly();
+    }
+
+
     public static int getFlag(Message message) {
         return message.getFlag();
     }