package com.liveramp.kafka_service.consumer.persist_helpers;

import java.util.Collection;
import java.util.Map;

import org.apache.kafka.common.TopicPartition;

import com.liveramp.kafka_service.zookeeper.ZKEnv;
import com.liveramp.kafka_service.zookeeper.ZookeeperClient;

public class ZookeeperPersistentHelper implements PersistentHelper {

  private final ZookeeperClient zookeeperClient;

  private ZookeeperPersistentHelper(final ZookeeperClient zookeeperClient) {
    this.zookeeperClient = zookeeperClient;
  }

  public static ZookeeperPersistentHelper createProductionHelper() {
    return new ZookeeperPersistentHelper(new ZookeeperClient.Builder(ZKEnv.PRODUCTION_ZKS).build());
  }

  public static ZookeeperPersistentHelper createTestHelper() {
    return new ZookeeperPersistentHelper(new ZookeeperClient.Builder(ZKEnv.TEST_ZKS).build());
  }

  @Override
  public boolean persistOffset(final TopicPartition partition, final long offset) {
    return false;
  }

  @Override
  public boolean persistOffsets(final Map<TopicPartition, Long> topicPartitionOffsets) {
    return false;
  }

  @Override
  public Long retrieveOffset(final TopicPartition partition) {
    return null;
  }

  @Override
  public Map<TopicPartition, Long> retrieveOffsets(final Collection<TopicPartition> partitions) {
    return null;
  }
}

@@ -963,137 +963,450 @@
 urn 
-false;%0A  %7D%0A%0A  @Override%0A  public boolean persistOffsets(final Map%3CTopicPartition, Long%3E topicPartitionOffsets) %7B%0A    return false
+zookeeperClient.deleteNode(partition.toString(), true) && zookeeperClient.createNode(partition.toString() + %22/%22 + offset);%0A  %7D%0A%0A  @Override%0A  public boolean persistOffsets(final Map%3CTopicPartition, Long%3E topicPartitionOffsets) %7B%0A    boolean success = true;%0A    for (Map.Entry%3CTopicPartition, Long%3E entry : topicPartitionOffsets.entrySet()) %7B%0A      success = success && persistOffset(entry.getKey(), entry.getValue());%0A    %7D%0A    return success
 ;%0A  

Implement persistOffset(s) for ZookeeperPersistentHelper