diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4b4bde6ad..58ec21217 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,13 @@ features that may not be compatible with old Kafka versions. If you don't
 specify this value it will default to 0.8.2 (the minimum supported), and trying
 to use more recent features (like the offset manager) will fail with an error.
 
+_Also:_ The offset-manager's behaviour has been changed to match the upstream
+java consumer (see [#705](https://github.com/Shopify/sarama/pull/705) and
+[#713](https://github.com/Shopify/sarama/pull/713)). If you use the
+offset-manager, please ensure that you are committing one *greater* than the
+last consumed message offset or else you may end up consuming duplicate
+messages.
+
 New Features:
  - Support for Kafka 0.10
    ([#672](https://github.com/Shopify/sarama/pull/672),
@@ -35,6 +42,8 @@ Bug Fixes:
    ([#685](https://github.com/Shopify/sarama/pull/685)).
  - Fix a possible tight loop in the consumer
    ([#693](https://github.com/Shopify/sarama/pull/693)).
+ - Match upstream's offset-tracking behaviour
+   ([#705](https://github.com/Shopify/sarama/pull/705)).
  - Report UnknownTopicOrPartition errors from the offset manager
    ([#706](https://github.com/Shopify/sarama/pull/706)).
  - Fix possible negative partition value from the HashPartitioner
diff --git a/functional_offset_manager_test.go b/functional_offset_manager_test.go
index b759669c0..436f35ef4 100644
--- a/functional_offset_manager_test.go
+++ b/functional_offset_manager_test.go
@@ -34,8 +34,8 @@ func TestFuncOffsetManager(t *testing.T) {
 
 	offset, metadata := pom2.NextOffset()
 
-	if offset != 10+1 {
-		t.Errorf("Expected the next offset to be 11, found %d.", offset)
+	if offset != 10 {
+		t.Errorf("Expected the next offset to be 10, found %d.", offset)
 	}
 	if metadata != "test metadata" {
 		t.Errorf("Expected metadata to be 'test metadata', found %s.", metadata)
diff --git a/offset_manager.go b/offset_manager.go
index 0c446b352..5e15cdafe 100644
--- a/offset_manager.go
+++ b/offset_manager.go
@@ -136,11 +136,15 @@ type PartitionOffsetManager interface {
 	// was committed for this partition yet.
 	NextOffset() (int64, string)
 
-	// MarkOffset marks the provided offset as processed, alongside a metadata string
+	// MarkOffset marks the provided offset, alongside a metadata string
 	// that represents the state of the partition consumer at that point in time. The
 	// metadata string can be used by another consumer to restore that state, so it
 	// can resume consumption.
 	//
+	// To follow upstream conventions, you are expected to mark the offset of the
+	// next message to read, not the last message read. Thus, when calling `MarkOffset`
+	// you should typically add one to the offset of the last consumed message.
+	//
 	// Note: calling MarkOffset does not necessarily commit the offset to the backend
 	// store immediately for efficiency reasons, and it may never be committed if
 	// your application crashes. This means that you may end up processing the same
@@ -340,7 +344,7 @@ func (pom *partitionOffsetManager) NextOffset() (int64, string) {
 	defer pom.lock.Unlock()
 
 	if pom.offset >= 0 {
-		return pom.offset + 1, pom.metadata
+		return pom.offset, pom.metadata
 	}
 
 	return pom.parent.conf.Consumer.Offsets.Initial, ""
diff --git a/offset_manager_test.go b/offset_manager_test.go
index 57008f9c1..c111a5a63 100644
--- a/offset_manager_test.go
+++ b/offset_manager_test.go
@@ -190,7 +190,7 @@ func TestPartitionOffsetManagerNextOffset(t *testing.T) {
 	pom := initPartitionOffsetManager(t, om, coordinator, 5, "test_meta")
 
 	offset, meta := pom.NextOffset()
-	if offset != 6 {
+	if offset != 5 {
 		t.Errorf("Expected offset 5. Actual: %v", offset)
 	}
 	if meta != "test_meta" {
@@ -215,7 +215,7 @@ func TestPartitionOffsetManagerMarkOffset(t *testing.T) {
 	pom.MarkOffset(100, "modified_meta")
 	offset, meta := pom.NextOffset()
 
-	if offset != 101 {
+	if offset != 100 {
 		t.Errorf("Expected offset 100. Actual: %v", offset)
 	}
 	if meta != "modified_meta" {
@@ -252,7 +252,7 @@ func TestPartitionOffsetManagerMarkOffsetWithRetention(t *testing.T) {
 	pom.MarkOffset(100, "modified_meta")
 	offset, meta := pom.NextOffset()
 
-	if offset != 101 {
+	if offset != 100 {
 		t.Errorf("Expected offset 100. Actual: %v", offset)
 	}
 	if meta != "modified_meta" {