text:
package org.apache.flink.runtime.state.filesystem;

import org.apache.flink.api.common.JobID;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.runtime.state.CheckpointStorageAccess;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory.FsCheckpointStateOutputStream;

import org.junit.Test;

import javax.annotation.Nonnull;

import java.io.File;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.Arrays;
import java.util.List;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;

/**
 * Tests for the {@link FsCheckpointStorageAccess}, which implements the checkpoint storage aspects
 * of the {@link FsStateBackend}.
 */
public class FsCheckpointStorageAccessTest extends AbstractFileCheckpointStorageAccessTestBase {

    private static final int FILE_SIZE_THRESHOLD = 1024;
    private static final int WRITE_BUFFER_SIZE = 4096;

    // ------------------------------------------------------------------------
    //  General Fs-based checkpoint storage tests, inherited
    // ------------------------------------------------------------------------

    @Override
    protected CheckpointStorageAccess createCheckpointStorage(Path checkpointDir) throws Exception {
        return new FsCheckpointStorageAccess(
                checkpointDir, null, new JobID(), FILE_SIZE_THRESHOLD, WRITE_BUFFER_SIZE);
    }

    @Override
    protected CheckpointStorageAccess createCheckpointStorageWithSavepointDir(
            Path checkpointDir, Path savepointDir) throws Exception {
        return new FsCheckpointStorageAccess(
                checkpointDir, savepointDir, new JobID(), FILE_SIZE_THRESHOLD, WRITE_BUFFER_SIZE);
    }

    // ------------------------------------------------------------------------
    //  FsCheckpointStorage-specific tests
    // ------------------------------------------------------------------------

    @Test
    public void testSavepointsInOneDirectoryDefaultLocation() throws Exception {
        final Path defaultSavepointDir = Path.fromLocalFile(tmp.newFolder());

        final FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        Path.fromLocalFile(tmp.newFolder()),
                        defaultSavepointDir,
                        new JobID(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        final FsCheckpointStorageLocation savepointLocation =
                (FsCheckpointStorageLocation) storage.initializeLocationForSavepoint(52452L, null);

        // all state types should be in the expected location
        assertParent(defaultSavepointDir, savepointLocation.getCheckpointDirectory());
        assertParent(defaultSavepointDir, savepointLocation.getSharedStateDirectory());
        assertParent(defaultSavepointDir, savepointLocation.getTaskOwnedStateDirectory());

        // cleanup
        savepointLocation.disposeOnFailure();
    }

    @Test
    public void testSavepointsInOneDirectoryCustomLocation() throws Exception {
        final Path savepointDir = Path.fromLocalFile(tmp.newFolder());

        final FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        Path.fromLocalFile(tmp.newFolder()),
                        null,
                        new JobID(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        final FsCheckpointStorageLocation savepointLocation =
                (FsCheckpointStorageLocation)
                        storage.initializeLocationForSavepoint(52452L, savepointDir.toString());

        // all state types should be in the expected location
        assertParent(savepointDir, savepointLocation.getCheckpointDirectory());
        assertParent(savepointDir, savepointLocation.getSharedStateDirectory());
        assertParent(savepointDir, savepointLocation.getTaskOwnedStateDirectory());

        // cleanup
        savepointLocation.disposeOnFailure();
    }

    @Test
    public void testTaskOwnedStateStream() throws Exception {
        final List<String> state = Arrays.asList("Flopsy", "Mopsy", "Cotton Tail", "Peter");

        // we chose a small size threshold here to force the state to disk
        final FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        Path.fromLocalFile(tmp.newFolder()),
                        null,
                        new JobID(),
                        10,
                        WRITE_BUFFER_SIZE);

        final StreamStateHandle stateHandle;

        try (CheckpointStateOutputStream stream = storage.createTaskOwnedStateStream()) {
            assertTrue(stream instanceof FsCheckpointStateOutputStream);

            new ObjectOutputStream(stream).writeObject(state);
            stateHandle = stream.closeAndGetHandle();
        }

        // the state must have gone to disk
        FileStateHandle fileStateHandle = (FileStateHandle) stateHandle;

        // check that the state is in the correct directory
        String parentDirName = fileStateHandle.getFilePath().getParent().getName();
        assertEquals(FsCheckpointStorageAccess.CHECKPOINT_TASK_OWNED_STATE_DIR, parentDirName);

        // validate the contents
        try (ObjectInputStream in = new ObjectInputStream(stateHandle.openInputStream())) {
            assertEquals(state, in.readObject());
        }
    }

    @Test
    public void testDirectoriesForExclusiveAndSharedState() throws Exception {
        final FileSystem fs = LocalFileSystem.getSharedInstance();
        final Path checkpointDir = randomTempPath();
        final Path sharedStateDir = randomTempPath();

        FsCheckpointStorageLocation storageLocation =
                new FsCheckpointStorageLocation(
                        fs,
                        checkpointDir,
                        sharedStateDir,
                        randomTempPath(),
                        CheckpointStorageLocationReference.getDefault(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        assertNotEquals(
                storageLocation.getCheckpointDirectory(),
                storageLocation.getSharedStateDirectory());

        assertEquals(0, fs.listStatus(storageLocation.getCheckpointDirectory()).length);
        assertEquals(0, fs.listStatus(storageLocation.getSharedStateDirectory()).length);

        // create exclusive state

        FsCheckpointStateOutputStream exclusiveStream =
                storageLocation.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);

        exclusiveStream.write(42);
        exclusiveStream.flushToFile();
        StreamStateHandle exclusiveHandle = exclusiveStream.closeAndGetHandle();

        assertEquals(1, fs.listStatus(storageLocation.getCheckpointDirectory()).length);
        assertEquals(0, fs.listStatus(storageLocation.getSharedStateDirectory()).length);

        // create shared state

        FsCheckpointStateOutputStream sharedStream =
                storageLocation.createCheckpointStateOutputStream(CheckpointedStateScope.SHARED);

        sharedStream.write(42);
        sharedStream.flushToFile();
        StreamStateHandle sharedHandle = sharedStream.closeAndGetHandle();

        assertEquals(1, fs.listStatus(storageLocation.getCheckpointDirectory()).length);
        assertEquals(1, fs.listStatus(storageLocation.getSharedStateDirectory()).length);

        // drop state

        exclusiveHandle.discardState();
        sharedHandle.discardState();
    }

    /**
     * This test checks that the expected mkdirs action for checkpoint storage, only be called when
     * initializeBaseLocations and not called when resolveCheckpointStorageLocation.
     */
    @Test
    public void testStorageLocationMkdirs() throws Exception {
        FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        randomTempPath(),
                        null,
                        new JobID(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        File baseDir = new File(storage.getCheckpointsDirectory().getPath());
        assertFalse(baseDir.exists());

        // mkdirs would only be called when initializeBaseLocations
        storage.initializeBaseLocationsForCheckpoint();
        assertTrue(baseDir.exists());

        // mkdir would not be called when resolveCheckpointStorageLocation
        storage =
                new FsCheckpointStorageAccess(
                        randomTempPath(),
                        null,
                        new JobID(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        FsCheckpointStorageLocation location =
                (FsCheckpointStorageLocation)
                        storage.resolveCheckpointStorageLocation(
                                177, CheckpointStorageLocationReference.getDefault());

        Path checkpointPath = location.getCheckpointDirectory();
        File checkpointDir = new File(checkpointPath.getPath());
        assertFalse(checkpointDir.exists());
    }

    @Test
    public void testResolveCheckpointStorageLocation() throws Exception {
        final FileSystem checkpointFileSystem = mock(FileSystem.class);
        final FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        new TestingPath("hdfs:///checkpoint/", checkpointFileSystem),
                        null,
                        new JobID(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        final FsCheckpointStorageLocation checkpointStreamFactory =
                (FsCheckpointStorageLocation)
                        storage.resolveCheckpointStorageLocation(
                                1L, CheckpointStorageLocationReference.getDefault());
        assertEquals(checkpointFileSystem, checkpointStreamFactory.getFileSystem());

        final CheckpointStorageLocationReference savepointLocationReference =
                AbstractFsCheckpointStorageAccess.encodePathAsReference(
                        new Path("file:///savepoint/"));

        final FsCheckpointStorageLocation savepointStreamFactory =
                (FsCheckpointStorageLocation)
                        storage.resolveCheckpointStorageLocation(2L, savepointLocationReference);
        final FileSystem fileSystem = savepointStreamFactory.getFileSystem();
        assertTrue(fileSystem instanceof LocalFileSystem);
    }

    // ------------------------------------------------------------------------
    //  Utilities
    // ------------------------------------------------------------------------

    private void assertParent(Path parent, Path child) {
        Path path = new Path(parent, child.getName());
        assertEquals(path, child);
    }

    private static final class TestingPath extends Path {

        private static final long serialVersionUID = 2560119808844230488L;

        @SuppressWarnings("TransientFieldNotInitialized")
        @Nonnull
        private final transient FileSystem fileSystem;

        TestingPath(String pathString, @Nonnull FileSystem fileSystem) {
            super(pathString);
            this.fileSystem = fileSystem;
        }

        @Override
        public FileSystem getFileSystem() {
            return fileSystem;
        }
    }
}


text:
![Build](https://github.com/bobbylight/ZeldaJS/actions/workflows/build.yml/badge.svg)
![CodeQL](https://github.com/bobbylight/ZeldaJS/actions/workflows/codeql-analysis.yml/badge.svg)
[![codecov](https://codecov.io/gh/bobbylight/ZeldaJS/branch/master/graph/badge.svg?token=IT4BJHTZKJ)](https://codecov.io/gh/bobbylight/ZeldaJS)
[![Dependency Status](https://img.shields.io/david/bobbylight/ZeldaJS.svg)](https://david-dm.org/bobbylight/ZeldaJS)
[![Dev Dependency Status](https://img.shields.io/david/dev/bobbylight/ZeldaJS.svg)](https://david-dm.org/bobbylight/ZeldaJS?type=dev)

A WIP clone of the Legend of Zelda.  This will include a very basic editor as well.
Feel free to [try it out in its current state](http://bobbylight.github.io/ZeldaJS/).

Licensed under [an MIT license](LICENSE.txt).

## Hacking
First, check out the project and install all dependencies locally via `npm`:

```bash
git clone https://github.com/bobbylight/ZeldaJS.git
cd ZeldaJS
npm install
```

To get started right away, run:

```bash
npm run serve
```

The game will be served from [http://localhost:8080](), and hot deploy any changes.

This game is built with `vue-cli`.  Development is done with `npm` scripts:

```bash
npm run clean          # Cleans build directories
npm run serve          # Runs app for development.  Hot deploys
npm run build          # Builds the application in dist/
npm run test:unit      # Runs unit tests and generates coverage
npm run lint           # Lints files
npm run doc            # Generates documentation
npm run build-electron # Doesn't currently work (broke with move to vue-cli)
npm run pack           # Doesn't currently work (broke with move to vue-cli)
npm run dist           # Doesn't currently work (broke with move to vue-cli)
```

The source code lives in `src/`, and is built into `dist/`.
`index.html` is the game itself, while `editor.html` is a simple map editor.

## Desktop Build
*Note:* These steps currently don't work.  Need to be fixed after the conversion
to vue-cli broke them.

A desktop build can be created:

```bash
npm run build-electron # Builds desktop resources into build/electron
npm run pack           # Creates unpacked game in dist/
npm run dist           # Creates unpacked game and installer in dist/
```

This is still a little rough around the edges, but works, at least on windows.

## Roadmap

### Implemented so far:

* Game
  - Map loading
  - Basic resource loading, Link sprite movement
  - Music
  - Moving between screens, smooth screen transitions (though only a portion of the map is implemented)
  - Entering and exiting caves
  - Use of sword (Z key)
  - Drop bombs, though they do nothing (X key)
  - Basic enemies that can die
  - Link takes damage from enemies and projectiles, and can die
  - Enemies drop rupees and hearts
* Editor
  - Screen design, single tile per cell (no burnable bushes, bombable walls, etc.)
  - Select enemies per screen
  - Enter and exit cave and level events

### Coming next (in no particular order):

* More accurate Link getting hit colors
* More accurate enemy spawning
* Expanded map
* Gibdos

redpajama_set_name: RedPajamaGithub


text:
namespace {

struct WaitableEventLazyInstanceTraits
    : public base::DefaultLazyInstanceTraits<base::WaitableEvent> {
  static base::WaitableEvent* New(void* instance) {
    // Use placement new to initialize our instance in our preallocated space.
    return new (instance) base::WaitableEvent(true, true);
  }
};

base::LazyInstance<base::WaitableEvent, WaitableEventLazyInstanceTraits>
    dummy_event = LAZY_INSTANCE_INITIALIZER;

base::StaticAtomicSequenceNumber g_next_id;

}  // namespace

namespace IPC {

#define kSyncMessageHeaderSize 4

SyncMessage::SyncMessage(
    int32 routing_id,
    uint32 type,
    PriorityValue priority,
    MessageReplyDeserializer* deserializer)
    : Message(routing_id, type, priority),
      deserializer_(deserializer),
      pump_messages_event_(NULL)
      {
  set_sync();
  set_unblock(true);

  // Add synchronous message data before the message payload.
  SyncHeader header;
  header.message_id = g_next_id.GetNext();
  WriteSyncHeader(this, header);
}

SyncMessage::~SyncMessage() {
}

MessageReplyDeserializer* SyncMessage::GetReplyDeserializer() {
  DCHECK(deserializer_.get());
  return deserializer_.release();
}

void SyncMessage::EnableMessagePumping() {
  DCHECK(!pump_messages_event_);
  set_pump_messages_event(dummy_event.Pointer());
}

bool SyncMessage::IsMessageReplyTo(const Message& msg, int request_id) {
  if (!msg.is_reply())
    return false;

  return GetMessageId(msg) == request_id;
}

PickleIterator SyncMessage::GetDataIterator(const Message* msg) {
  PickleIterator iter(*msg);
  if (!iter.SkipBytes(kSyncMessageHeaderSize))
    return PickleIterator();
  else
    return iter;
}

int SyncMessage::GetMessageId(const Message& msg) {
  if (!msg.is_sync() && !msg.is_reply())
    return 0;

  SyncHeader header;
  if (!ReadSyncHeader(msg, &header))
    return 0;

  return header.message_id;
}

Message* SyncMessage::GenerateReply(const Message* msg) {
  DCHECK(msg->is_sync());

  Message* reply = new Message(msg->routing_id(), IPC_REPLY_ID,
                               msg->priority());
  reply->set_reply();

  SyncHeader header;

  // use the same message id, but this time reply bit is set
  header.message_id = GetMessageId(*msg);
  WriteSyncHeader(reply, header);

  return reply;
}

bool SyncMessage::ReadSyncHeader(const Message& msg, SyncHeader* header) {
  DCHECK(msg.is_sync() || msg.is_reply());

  PickleIterator iter(msg);
  bool result = iter.ReadInt(&header->message_id);
  if (!result) {
    NOTREACHED();
    return false;
  }

  return true;
}

bool SyncMessage::WriteSyncHeader(Message* msg, const SyncHeader& header) {
  DCHECK(msg->is_sync() || msg->is_reply());
  DCHECK(msg->payload_size() == 0);
  bool result = msg->WriteInt(header.message_id);
  if (!result) {
    NOTREACHED();
    return false;
  }

  // Note: if you add anything here, you need to update kSyncMessageHeaderSize.
  DCHECK(kSyncMessageHeaderSize == msg->payload_size());

  return true;
}

bool MessageReplyDeserializer::SerializeOutputParameters(const Message& msg) {
  return SerializeOutputParameters(msg, SyncMessage::GetDataIterator(&msg));
}

}  // namespace IPC


text:
class ProjectApprovalsController < ApplicationController
  before_action :pre_hook
  after_action :post_hook

  api :GET, '/projects/:project_id/approvals', 'Returns a list of all approvals for a project'
  param :project_id, :number, required: true

  def index
    respond_with project.approvals
  end

  api :POST, '/projects/:id/approve', 'Set or change the approval for current_user for a project'
  param :project_id, :number, required: true
  param :includes, Array, required: false, in: (ProjectsController::PROJECT_INCLUDES - ['staff'])

  def update
    approval = project.approvals.find_or_initialize_by(staff_id: current_user.id)
    perform_in_transaction(project) { approval.approve! }
  end

  api :DELETE, '/projects/:project_id/reject', 'Set or change the approval for current_user for a project'
  param :project_id, :number, required: true
  param :includes, Array, required: false, in: (ProjectsController::PROJECT_INCLUDES - ['staff'])
  param :reason, String, required: true
  error code: 422, desc: ParameterValidation::Messages.missing

  def destroy
    approval = project.approvals.find_or_initialize_by(staff_id: current_user.id)
    perform_in_transaction(project) { approval.reject!(params[:reason]) }
  end

  private

  def perform_in_transaction(project)
    yield
    respond_with_params project
  rescue ActiveRecord::RecordInvalid => ex
    respond_with ex.record
  end

  def project
    @_project ||= Project.find(params[:project_id]).tap { |proj| authorize(proj) }
  end
end


text:
package com.linkedin.pinot.controller.helix.core.realtime;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Uninterruptibles;
import com.linkedin.pinot.common.config.ColumnPartitionConfig;
import com.linkedin.pinot.common.config.SegmentPartitionConfig;
import com.linkedin.pinot.common.config.TableConfig;
import com.linkedin.pinot.common.config.TableNameBuilder;
import com.linkedin.pinot.common.exception.InvalidConfigException;
import com.linkedin.pinot.common.metadata.ZKMetadataProvider;
import com.linkedin.pinot.common.metadata.segment.ColumnPartitionMetadata;
import com.linkedin.pinot.common.metadata.segment.LLCRealtimeSegmentZKMetadata;
import com.linkedin.pinot.common.metadata.segment.SegmentPartitionMetadata;
import com.linkedin.pinot.common.metrics.ControllerMeter;
import com.linkedin.pinot.common.metrics.ControllerMetrics;
import com.linkedin.pinot.common.partition.PartitionAssignment;
import com.linkedin.pinot.common.partition.StreamPartitionAssignmentGenerator;
import com.linkedin.pinot.common.protocols.SegmentCompletionProtocol;
import com.linkedin.pinot.common.utils.CommonConstants;
import com.linkedin.pinot.common.utils.LLCSegmentName;
import com.linkedin.pinot.common.utils.SegmentName;
import com.linkedin.pinot.common.utils.StringUtil;
import com.linkedin.pinot.common.utils.TarGzCompressionUtils;
import com.linkedin.pinot.common.utils.helix.HelixHelper;
import com.linkedin.pinot.common.utils.retry.RetryPolicies;
import com.linkedin.pinot.controller.ControllerConf;
import com.linkedin.pinot.controller.api.events.MetadataEventNotifierFactory;
import com.linkedin.pinot.controller.helix.core.PinotHelixResourceManager;
import com.linkedin.pinot.controller.helix.core.PinotHelixSegmentOnlineOfflineStateModelGenerator;
import com.linkedin.pinot.controller.helix.core.PinotTableIdealStateBuilder;
import com.linkedin.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
import com.linkedin.pinot.controller.helix.core.realtime.segment.FlushThresholdUpdateManager;
import com.linkedin.pinot.controller.helix.core.realtime.segment.FlushThresholdUpdater;
import com.linkedin.pinot.controller.util.SegmentCompletionUtils;
import com.linkedin.pinot.core.realtime.segment.ConsumingSegmentAssignmentStrategy;
import com.linkedin.pinot.core.realtime.segment.RealtimeSegmentAssignmentStrategy;
import com.linkedin.pinot.core.realtime.stream.OffsetCriteria;
import com.linkedin.pinot.core.realtime.stream.PartitionOffsetFetcher;
import com.linkedin.pinot.core.realtime.stream.StreamConfig;
import com.linkedin.pinot.core.realtime.stream.StreamConfigProperties;
import com.linkedin.pinot.core.segment.creator.impl.V1Constants;
import com.linkedin.pinot.core.segment.index.ColumnMetadata;
import com.linkedin.pinot.core.segment.index.SegmentMetadataImpl;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.math.IntRange;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixManager;
import org.apache.helix.ZNRecord;
import org.apache.helix.model.IdealState;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PinotLLCRealtimeSegmentManager {
  public static final Logger LOGGER = LoggerFactory.getLogger(PinotLLCRealtimeSegmentManager.class);
  protected static final int STARTING_SEQUENCE_NUMBER = 0; // Initial sequence number for new table segments
  protected static final long END_OFFSET_FOR_CONSUMING_SEGMENTS = Long.MAX_VALUE;

  private static final String METADATA_TEMP_DIR_SUFFIX = ".metadata.tmp";
  private static final String METADATA_EVENT_NOTIFIER_PREFIX = "metadata.event.notifier";
  // TODO: make this configurable with default set to 10
  /**
   * After step 1 of segment completion is done,
   * this is the max time until which step 3 is allowed to complete.
   * See {@link PinotLLCRealtimeSegmentManager#commitSegmentMetadata(String, CommittingSegmentDescriptor)} for explanation of steps 1 2 3
   * This includes any backoffs and retries for the steps 2 and 3
   * The segment will be eligible for repairs by the validation manager, if the time  exceeds this value
   */
  private static int MAX_SEGMENT_COMPLETION_TIME_MINS = 10;

  private static PinotLLCRealtimeSegmentManager INSTANCE = null;

  private final HelixAdmin _helixAdmin;
  private final HelixManager _helixManager;
  private final ZkHelixPropertyStore<ZNRecord> _propertyStore;
  private final PinotHelixResourceManager _helixResourceManager;
  private final String _clusterName;
  private boolean _amILeader = false;
  private final ControllerConf _controllerConf;
  private final ControllerMetrics _controllerMetrics;
  private final int _numIdealStateUpdateLocks;
  private final Lock[] _idealstateUpdateLocks;
  private final TableConfigCache _tableConfigCache;
  private final StreamPartitionAssignmentGenerator _streamPartitionAssignmentGenerator;
  private final FlushThresholdUpdateManager _flushThresholdUpdateManager;

  public boolean getIsSplitCommitEnabled() {
    return _controllerConf.getAcceptSplitCommit();
  }

  public String getControllerVipUrl() {
    return _controllerConf.generateVipUrl();
  }

  public static synchronized void create(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics) {
    create(helixResourceManager.getHelixAdmin(), helixResourceManager.getHelixClusterName(),
        helixResourceManager.getHelixZkManager(), helixResourceManager.getPropertyStore(), helixResourceManager,
        controllerConf, controllerMetrics);
  }

  private static synchronized void create(HelixAdmin helixAdmin, String clusterName, HelixManager helixManager,
      ZkHelixPropertyStore propertyStore, PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics) {
    if (INSTANCE != null) {
      throw new RuntimeException("Instance already created");
    }
    INSTANCE =
        new PinotLLCRealtimeSegmentManager(helixAdmin, clusterName, helixManager, propertyStore, helixResourceManager,
            controllerConf, controllerMetrics);
    SegmentCompletionManager.create(helixManager, INSTANCE, controllerConf, controllerMetrics);
  }

  public void start() {
    _helixManager.addControllerListener(changeContext -> onBecomeLeader());
  }

  protected PinotLLCRealtimeSegmentManager(HelixAdmin helixAdmin, String clusterName, HelixManager helixManager,
      ZkHelixPropertyStore propertyStore, PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics) {
    _helixAdmin = helixAdmin;
    _helixManager = helixManager;
    _propertyStore = propertyStore;
    _helixResourceManager = helixResourceManager;
    _clusterName = clusterName;
    _controllerConf = controllerConf;
    _controllerMetrics = controllerMetrics;
    _numIdealStateUpdateLocks = controllerConf.getRealtimeSegmentMetadataCommitNumLocks();
    _idealstateUpdateLocks = new Lock[_numIdealStateUpdateLocks];
    for (int i = 0; i < _numIdealStateUpdateLocks; i++) {
      _idealstateUpdateLocks[i] = new ReentrantLock();
    }
    _tableConfigCache = new TableConfigCache(_propertyStore);
    _streamPartitionAssignmentGenerator = new StreamPartitionAssignmentGenerator(_helixManager);
    _flushThresholdUpdateManager = new FlushThresholdUpdateManager();
  }

  public static PinotLLCRealtimeSegmentManager getInstance() {
    if (INSTANCE == null) {
      throw new RuntimeException("Not yet created");
    }
    return INSTANCE;
  }

  private void onBecomeLeader() {
    if (isLeader()) {
      if (!_amILeader) {
        // We were not leader before, now we are.
        _amILeader = true;
        LOGGER.info("Became leader");
      } else {
        // We already had leadership, nothing to do.
        LOGGER.info("Already leader. Duplicate notification");
      }
    } else {
      _amILeader = false;
      LOGGER.info("Lost leadership");
    }
  }

  protected boolean isLeader() {
    return _helixManager.isLeader();
  }

  protected boolean isConnected() {
    return _helixManager.isConnected();
  }

  /**
   *
   * @param tableConfig
   * @param emptyIdealState may contain HLC segments if both HLC and LLC are configured
   */
  public void setupNewTable(TableConfig tableConfig, IdealState emptyIdealState) throws InvalidConfigException {
    final StreamConfig streamConfig = new StreamConfig(tableConfig.getIndexingConfig().getStreamConfigs());
    int partitionCount = getPartitionCount(streamConfig);
    List<String> currentSegments = getExistingSegments(tableConfig.getTableName());
    // Make sure that there are no low-level segments existing.
    if (currentSegments != null) {
      for (String segment : currentSegments) {
        if (!SegmentName.isHighLevelConsumerSegmentName(segment)) {
          // For now, we don't support re-creating the low-level realtime segments
          throw new RuntimeException("Low-level segments already exist for table " + tableConfig.getTableType());
        }
      }
    }
    _flushThresholdUpdateManager.clearFlushThresholdUpdater(tableConfig);
    if (!isConnected()) {
      throw new RuntimeException(
          "Lost zk connection while setting up new table " + tableConfig.getTableName() + " isConnected="
              + isConnected());
    }
    IdealState idealState = setupTable(tableConfig, emptyIdealState, partitionCount);
    setTableIdealState(tableConfig.getTableName(), idealState);
  }

  // Remove all trace of LLC for this table.
  public void cleanupLLC(final String realtimeTableName) {
    // If there are any completions in the pipeline we let them commit.
    Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);

    IdealState idealState = HelixHelper.getTableIdealState(_helixManager, realtimeTableName);
    final List<String> segmentsToRemove = new ArrayList<String>();
    Set<String> allSegments = idealState.getPartitionSet();
    int removeCount = 0;
    for (String segmentName : allSegments) {
      if (SegmentName.isLowLevelConsumerSegmentName(segmentName)) {
        segmentsToRemove.add(segmentName);
        removeCount++;
      }
    }
    LOGGER.info("Attempting to remove {} LLC segments of table {}", removeCount, realtimeTableName);

    _helixResourceManager.deleteSegments(realtimeTableName, segmentsToRemove);
  }

  private SegmentPartitionMetadata getPartitionMetadataFromTableConfig(String tableName, int numPartitions,
      int partitionId) {
    Map<String, ColumnPartitionMetadata> partitionMetadataMap = new HashMap<>();
    if (_propertyStore == null) {
      return null;
    }
    TableConfig tableConfig = getRealtimeTableConfig(tableName);
    SegmentPartitionMetadata partitionMetadata = null;
    SegmentPartitionConfig partitionConfig = tableConfig.getIndexingConfig().getSegmentPartitionConfig();
    if (partitionConfig != null && partitionConfig.getColumnPartitionMap() != null
        && partitionConfig.getColumnPartitionMap().size() > 0) {
      Map<String, ColumnPartitionConfig> columnPartitionMap = partitionConfig.getColumnPartitionMap();
      for (Map.Entry<String, ColumnPartitionConfig> entry : columnPartitionMap.entrySet()) {
        String column = entry.getKey();
        ColumnPartitionConfig columnPartitionConfig = entry.getValue();
        partitionMetadataMap.put(column,
            new ColumnPartitionMetadata(columnPartitionConfig.getFunctionName(), numPartitions,
                Collections.singletonList(new IntRange(partitionId))));
      }
      partitionMetadata = new SegmentPartitionMetadata(partitionMetadataMap);
    }
    return partitionMetadata;
  }

  /**
   * Given a segment metadata, build the finalized version of segment partition metadata. This partition metadata will
   * be written as a part of SegmentZKMetadata.
   * @param segmentMetadata Segment metadata
   * @return
   */
  private SegmentPartitionMetadata getPartitionMetadataFromSegmentMetadata(SegmentMetadataImpl segmentMetadata) {
    Map<String, ColumnPartitionMetadata> partitionMetadataMap = new HashMap<>();
    for (Map.Entry<String, ColumnMetadata> entry : segmentMetadata.getColumnMetadataMap().entrySet()) {
      String columnName = entry.getKey();
      ColumnMetadata columnMetadata = entry.getValue();
      // Check if the column metadata contains the partition information
      if (columnMetadata.getPartitionFunction() != null && columnMetadata.getPartitionRanges() != null) {
        partitionMetadataMap.put(columnName,
            new ColumnPartitionMetadata(columnMetadata.getPartitionFunction().toString(),
                columnMetadata.getNumPartitions(), columnMetadata.getPartitionRanges()));
      }
    }
    return new SegmentPartitionMetadata(partitionMetadataMap);
  }

  protected List<String> getExistingSegments(String realtimeTableName) {
    String propStorePath = ZKMetadataProvider.constructPropertyStorePathForResource(realtimeTableName);
    return _propertyStore.getChildNames(propStorePath, AccessOption.PERSISTENT);
  }

  protected boolean writeSegmentToPropertyStore(String znodePath, ZNRecord znRecord, final String realtimeTableName,
      int expectedVersion) {
    boolean success = _propertyStore.set(znodePath, znRecord, expectedVersion, AccessOption.PERSISTENT);
    if (!success) {
      LOGGER.error(
          "Failed to write segment to property store at {} for table {}. Expected zookeeper version number: {}",
          znodePath, realtimeTableName, expectedVersion);
      return false;
    }
    return success;
  }

  protected boolean writeSegmentToPropertyStore(String znodePath, ZNRecord znRecord, final String realtimeTableName) {
    boolean success = _propertyStore.set(znodePath, znRecord, AccessOption.PERSISTENT);
    if (!success) {
      LOGGER.error("Failed to write segment to property store at {} for table {}.", znodePath, realtimeTableName);
    }
    return success;
  }

  protected void writeSegmentsToPropertyStore(List<String> paths, List<ZNRecord> records,
      final String realtimeTableName) {
    try {
      _propertyStore.setChildren(paths, records, AccessOption.PERSISTENT);
    } catch (Exception e) {
      LOGGER.error("Failed to update idealstate for table {} for paths {}", realtimeTableName, paths, e);
      _controllerMetrics.addMeteredGlobalValue(ControllerMeter.LLC_ZOOKEEPER_UPDATE_FAILURES, 1);
      throw e;
    }
  }

  protected IdealState getTableIdealState(String realtimeTableName) {
    return HelixHelper.getTableIdealState(_helixManager, realtimeTableName);
  }

  protected void setTableIdealState(String realtimeTableName, IdealState idealState) {
    _helixAdmin.setResourceIdealState(_clusterName, realtimeTableName, idealState);
  }

  public boolean commitSegmentFile(String tableName, CommittingSegmentDescriptor committingSegmentDescriptor) {
    String segmentName = committingSegmentDescriptor.getSegmentName();
    String segmentLocation = committingSegmentDescriptor.getSegmentLocation();
    File segmentFile = convertURIToSegmentLocation(segmentLocation);

    File baseDir = new File(_controllerConf.getDataDir());
    File tableDir = new File(baseDir, tableName);
    File fileToMoveTo = new File(tableDir, segmentName);

    if (!isConnected() || !isLeader()) {
      // We can potentially log a different value than what we saw ....
      LOGGER.warn("Lost leadership while committing segment file {}, {} for table {}: isLeader={}, isConnected={}",
          segmentName, segmentLocation, tableName, isLeader(), isConnected());
      _controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
      return false;
    }

    try {
      com.linkedin.pinot.common.utils.FileUtils.moveFileWithOverwrite(segmentFile, fileToMoveTo);
    } catch (Exception e) {
      LOGGER.error("Could not move {} to {}", segmentFile, segmentName, e);
      return false;
    }
    for (File file : tableDir.listFiles()) {
      if (file.getName().startsWith(SegmentCompletionUtils.getSegmentNamePrefix(segmentName))) {
        LOGGER.warn("Deleting " + file);
        FileUtils.deleteQuietly(file);
      }
    }
    return true;
  }

  private static File convertURIToSegmentLocation(String segmentLocation) {
    try {
      URI uri = new URI(segmentLocation);
      return new File(uri.getPath());
    } catch (URISyntaxException e) {
      throw new RuntimeException("Could not convert URI " + segmentLocation + " to segment location", e);
    }
  }

  /**
   * This method is invoked after the realtime segment is uploaded but before a response is sent to the server.
   * It updates the propertystore segment metadata from IN_PROGRESS to DONE, and also creates new propertystore
   * records for new segments, and puts them in idealstate in CONSUMING state.
   *
   * @param rawTableName Raw table name
   * @param committingSegmentDescriptor
   * @return boolean
   */
  public boolean commitSegmentMetadata(String rawTableName, CommittingSegmentDescriptor committingSegmentDescriptor) {
    final String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
    TableConfig tableConfig = getRealtimeTableConfig(realtimeTableName);
    if (tableConfig == null) {
      LOGGER.warn("Did not find table config for table {}", realtimeTableName);
      return false;
    }

    /*
     * Update zookeeper in 3 steps.
     *
     * Step 1: Update PROPERTYSTORE to change the old segment metadata status to DONE
     * Step 2: Update PROPERTYSTORE to create the new segment metadata with status IN_PROGRESS
     * Step 3: Update IDEALSTATES to include new segment in CONSUMING state, and change old segment to ONLINE state.
     */

    final String committingSegmentNameStr = committingSegmentDescriptor.getSegmentName();
    final long nextOffset = committingSegmentDescriptor.getNextOffset();
    final LLCSegmentName committingLLCSegmentName = new LLCSegmentName(committingSegmentNameStr);
    final int partitionId = committingLLCSegmentName.getPartitionId();
    final int newSeqNum = committingLLCSegmentName.getSequenceNumber() + 1;
    final long now = System.currentTimeMillis();

    LLCSegmentName newLLCSegmentName =
        new LLCSegmentName(committingLLCSegmentName.getTableName(), partitionId, newSeqNum, now);
    String newSegmentNameStr = newLLCSegmentName.getSegmentName();

    IdealState idealState = getTableIdealState(realtimeTableName);
    Preconditions.checkState(idealState.getInstanceStateMap(committingSegmentNameStr)
        .containsValue(PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE));
    int numPartitions = _streamPartitionAssignmentGenerator.getNumPartitionsFromIdealState(idealState);

    PartitionAssignment partitionAssignment;
    try {
      partitionAssignment =
          _streamPartitionAssignmentGenerator.generateStreamPartitionAssignment(tableConfig, numPartitions);
    } catch (InvalidConfigException e) {
      LOGGER.error("Exception when generating partition assignment for table {} and numPartitions {}",
          realtimeTableName, numPartitions, e);
      return false;
    }

    // If an LLC table is dropped (or cleaned up), we will get null here. In that case we should not be creating a new segment
    if (partitionAssignment == null) {
      LOGGER.warn("Partition assignment not found for {}", realtimeTableName);
      throw new RuntimeException("Partition assignment not found. Not committing segment");
    }

    // Step-1
    boolean success = updateOldSegmentMetadataZNRecord(realtimeTableName, committingLLCSegmentName, nextOffset);
    if (!success) {
      return false;
    }

    // Step-2
    success =
        createNewSegmentMetadataZNRecord(tableConfig, committingLLCSegmentName, newLLCSegmentName, partitionAssignment,
            committingSegmentDescriptor, false);
    if (!success) {
      return false;
    }

    // Step-3
    // TODO Introduce a controller failure here for integration testing

    // When multiple segments of the same table complete around the same time it is possible that
    // the idealstate udpate fails due to contention. We serialize the updates to the idealstate
    // to reduce this contention. We may still contend with RetentionManager, or other updates
    // to idealstate from other controllers, but then we have the retry mechanism to get around that.
    // hash code can be negative, so make sure we are getting a positive lock index
    int lockIndex = (realtimeTableName.hashCode() & Integer.MAX_VALUE) % _numIdealStateUpdateLocks;
    Lock lock = _idealstateUpdateLocks[lockIndex];
    try {
      lock.lock();
      updateIdealStateOnSegmentCompletion(realtimeTableName, committingSegmentNameStr, newSegmentNameStr,
          partitionAssignment);
      LOGGER.info("Changed {} to ONLINE and created {} in CONSUMING", committingSegmentNameStr, newSegmentNameStr);
    } finally {
      lock.unlock();
    }

    // Trigger the metadata event notifier
    notifyOnSegmentFlush(realtimeTableName);

    return true;
  }

  /**
   * Update segment metadata of committing segment
   * @param realtimeTableName - table name for which segment is being committed
   * @param committingLLCSegmentName - name of the segment being committed
   * @param nextOffset - the end offset for this committing segment
   * @return
   */
  protected boolean updateOldSegmentMetadataZNRecord(String realtimeTableName, LLCSegmentName committingLLCSegmentName,
      long nextOffset) {

    String committingSegmentNameStr = committingLLCSegmentName.getSegmentName();
    Stat stat = new Stat();
    final LLCRealtimeSegmentZKMetadata committingSegmentMetadata =
        getRealtimeSegmentZKMetadata(realtimeTableName, committingSegmentNameStr, stat);

    if (committingSegmentMetadata.getStatus() != CommonConstants.Segment.Realtime.Status.IN_PROGRESS) {
      LOGGER.warn("Status of segment metadata {} has already been changed by other controller for table {}: Status={}",
          committingSegmentNameStr, realtimeTableName, committingSegmentMetadata.getStatus());
      return false;
    }

    // TODO: set number of rows to end consumption in new segment metadata, based on memory used and number of rows from old segment
    committingSegmentMetadata.setEndOffset(nextOffset);
    committingSegmentMetadata.setStatus(CommonConstants.Segment.Realtime.Status.DONE);
    String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
    committingSegmentMetadata.setDownloadUrl(
        ControllerConf.constructDownloadUrl(rawTableName, committingSegmentNameStr, _controllerConf.generateVipUrl()));
    // Pull segment metadata from incoming segment and set it in zk segment metadata
    SegmentMetadataImpl segmentMetadata = extractSegmentMetadata(rawTableName, committingSegmentNameStr);
    committingSegmentMetadata.setCrc(Long.valueOf(segmentMetadata.getCrc()));
    committingSegmentMetadata.setStartTime(segmentMetadata.getTimeInterval().getStartMillis());
    committingSegmentMetadata.setEndTime(segmentMetadata.getTimeInterval().getEndMillis());
    committingSegmentMetadata.setTimeUnit(TimeUnit.MILLISECONDS);
    committingSegmentMetadata.setIndexVersion(segmentMetadata.getVersion());
    committingSegmentMetadata.setTotalRawDocs(segmentMetadata.getTotalRawDocs());
    committingSegmentMetadata.setPartitionMetadata(getPartitionMetadataFromSegmentMetadata(segmentMetadata));

    final ZNRecord oldZnRecord = committingSegmentMetadata.toZNRecord();
    final String oldZnodePath =
        ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, committingSegmentNameStr);

    if (!isConnected() || !isLeader()) {
      // We can potentially log a different value than what we saw ....
      LOGGER.warn("Lost leadership while committing segment metadata for {} for table {}: isLeader={}, isConnected={}",
          committingSegmentNameStr, realtimeTableName, isLeader(), isConnected());
      _controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
      return false;
    }
    boolean success = writeSegmentToPropertyStore(oldZnodePath, oldZnRecord, realtimeTableName, stat.getVersion());
    if (!success) {
      LOGGER.warn("Fail to write old segment to property store for {} for table {}: isLeader={}, isConnected={}",
          committingSegmentNameStr, realtimeTableName, isLeader(), isConnected());
    }
    return success;
  }

  /**
   * Creates segment metadata for next sequence number from the segment just committed
   * @param realtimeTableConfig  table config of the segment for which new metadata is being created
   * @param committingLLCSegmentName
   * @param newLLCSegmentName  new segment name
   * @param partitionAssignment  stream partition assignment for this table
   * @param committingSegmentDescriptor
   * @param isNewTableSetup
   * @return
   */
  protected boolean createNewSegmentMetadataZNRecord(TableConfig realtimeTableConfig,
      LLCSegmentName committingLLCSegmentName, LLCSegmentName newLLCSegmentName,
      PartitionAssignment partitionAssignment, CommittingSegmentDescriptor committingSegmentDescriptor,
      boolean isNewTableSetup) {

    String realtimeTableName = realtimeTableConfig.getTableName();
    String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);

    LLCRealtimeSegmentZKMetadata committingSegmentZKMetadata = null;
    if (committingLLCSegmentName != null) {
      committingSegmentZKMetadata =
          getRealtimeSegmentZKMetadata(realtimeTableName, committingLLCSegmentName.getSegmentName(), null);
    }

    int partitionId = newLLCSegmentName.getPartitionId();
    int numReplicas = partitionAssignment.getInstancesListForPartition(String.valueOf(partitionId)).size();
    ZNRecord newZnRecord =
        makeZnRecordForNewSegment(rawTableName, numReplicas, committingSegmentDescriptor.getNextOffset(),
            newLLCSegmentName, partitionAssignment.getNumPartitions());
    final LLCRealtimeSegmentZKMetadata newSegmentZKMetadata = new LLCRealtimeSegmentZKMetadata(newZnRecord);

    FlushThresholdUpdater flushThresholdUpdater =
        _flushThresholdUpdateManager.getFlushThresholdUpdater(realtimeTableConfig);
    flushThresholdUpdater.updateFlushThreshold(newSegmentZKMetadata, committingSegmentZKMetadata,
        committingSegmentDescriptor, partitionAssignment);

    newZnRecord = newSegmentZKMetadata.toZNRecord();

    final String newSegmentNameStr = newLLCSegmentName.getSegmentName();
    final String newZnodePath =
        ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, newSegmentNameStr);

    if (!isNewTableSetup) {
      if (!isLeader() || !isConnected()) {
        // We can potentially log a different value than what we saw ....
        LOGGER.warn(
            "Lost leadership while committing new segment metadata for {} for table {}: isLeader={}, isConnected={}",
            newSegmentNameStr, rawTableName, isLeader(), isConnected());
        _controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
        return false;
      }
    }

    boolean success = writeSegmentToPropertyStore(newZnodePath, newZnRecord, realtimeTableName);
    if (!success) {
      LOGGER.warn("Fail to write new segment to property store for {} for table {}: isLeader={}, isConnected={}",
          newSegmentNameStr, rawTableName, isLeader(), isConnected());
    }
    return success;
  }

  /**
   * Helper function to return cached table config.
   *
   * @param tableName name of the table
   * @return table configuration that reflects the most recent version
   */
  protected TableConfig getRealtimeTableConfig(String tableName) {
    TableConfig tableConfig;
    String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
    try {
      tableConfig = _tableConfigCache.getTableConfig(tableNameWithType);
    } catch (ExecutionException e) {
      LOGGER.warn("Exception happened while loading the table config ({}) from the property store to the cache.",
          tableNameWithType, e);
      return null;
    }
    return tableConfig;
  }

  public long getCommitTimeoutMS(String tableName) {
    long commitTimeoutMS = SegmentCompletionProtocol.getMaxSegmentCommitTimeMs();
    if (_propertyStore == null) {
      return commitTimeoutMS;
    }
    TableConfig tableConfig = getRealtimeTableConfig(tableName);
    final Map<String, String> streamConfigs = tableConfig.getIndexingConfig().getStreamConfigs();
    if (streamConfigs != null && streamConfigs.containsKey(StreamConfigProperties.SEGMENT_COMMIT_TIMEOUT_SECONDS)) {
      final String commitTimeoutSecondsStr = streamConfigs.get(StreamConfigProperties.SEGMENT_COMMIT_TIMEOUT_SECONDS);
      try {
        return TimeUnit.MILLISECONDS.convert(Integer.parseInt(commitTimeoutSecondsStr), TimeUnit.SECONDS);
      } catch (Exception e) {
        LOGGER.warn("Failed to parse flush size of {}", commitTimeoutSecondsStr, e);
        return commitTimeoutMS;
      }
    }
    return commitTimeoutMS;
  }

  /**
   * Extract the segment metadata files from the tar-zipped segment file that is expected to be in the directory for the
   * table.
   * <p>Segment tar-zipped file path: DATADIR/rawTableName/segmentName.
   * <p>We extract the metadata.properties and creation.meta into a temporary metadata directory:
   * DATADIR/rawTableName/segmentName.metadata.tmp, and load metadata from there.
   *
   * @param rawTableName Name of the table (not including the REALTIME extension)
   * @param segmentNameStr Name of the segment
   * @return SegmentMetadataImpl if it is able to extract the metadata file from the tar-zipped segment file.
   */
  protected SegmentMetadataImpl extractSegmentMetadata(final String rawTableName, final String segmentNameStr) {
    String baseDirStr = StringUtil.join("/", _controllerConf.getDataDir(), rawTableName);
    String segFileStr = StringUtil.join("/", baseDirStr, segmentNameStr);
    String tempMetadataDirStr = StringUtil.join("/", baseDirStr, segmentNameStr + METADATA_TEMP_DIR_SUFFIX);
    File tempMetadataDir = new File(tempMetadataDirStr);

    try {
      Preconditions.checkState(tempMetadataDir.mkdirs(), "Failed to create directory: %s", tempMetadataDirStr);

      // Extract metadata.properties
      InputStream metadataPropertiesInputStream =
          TarGzCompressionUtils.unTarOneFile(new FileInputStream(new File(segFileStr)),
              V1Constants.MetadataKeys.METADATA_FILE_NAME);
      Preconditions.checkNotNull(metadataPropertiesInputStream, "%s does not exist",
          V1Constants.MetadataKeys.METADATA_FILE_NAME);
      Path metadataPropertiesPath =
          FileSystems.getDefault().getPath(tempMetadataDirStr, V1Constants.MetadataKeys.METADATA_FILE_NAME);
      Files.copy(metadataPropertiesInputStream, metadataPropertiesPath);

      // Extract creation.meta
      InputStream creationMetaInputStream =
          TarGzCompressionUtils.unTarOneFile(new FileInputStream(new File(segFileStr)),
              V1Constants.SEGMENT_CREATION_META);
      Preconditions.checkNotNull(creationMetaInputStream, "%s does not exist", V1Constants.SEGMENT_CREATION_META);
      Path creationMetaPath = FileSystems.getDefault().getPath(tempMetadataDirStr, V1Constants.SEGMENT_CREATION_META);
      Files.copy(creationMetaInputStream, creationMetaPath);

      // Load segment metadata
      return new SegmentMetadataImpl(tempMetadataDir);
    } catch (Exception e) {
      throw new RuntimeException("Exception extracting and reading segment metadata for " + segmentNameStr, e);
    } finally {
      FileUtils.deleteQuietly(tempMetadataDir);
    }
  }

  public LLCRealtimeSegmentZKMetadata getRealtimeSegmentZKMetadata(String realtimeTableName, String segmentName,
      Stat stat) {
    ZNRecord znRecord =
        _propertyStore.get(ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, segmentName),
            stat, AccessOption.PERSISTENT);
    if (znRecord == null) {
      LOGGER.error("Segment metadata not found for table {}, segment {}. (can happen during table drop)",
          realtimeTableName, segmentName);
      throw new RuntimeException(
          "Segment metadata not found for table " + realtimeTableName + " segment " + segmentName);
    }
    return new LLCRealtimeSegmentZKMetadata(znRecord);
  }

  protected long getPartitionOffset(StreamConfig streamConfig, final OffsetCriteria offsetCriteria,
      int partitionId) {
    return fetchPartitionOffset(streamConfig, offsetCriteria, partitionId);
  }

  private long fetchPartitionOffset(StreamConfig streamConfig, final OffsetCriteria offsetCriteria, int partitionId) {
    PartitionOffsetFetcher partitionOffsetFetcher =
        new PartitionOffsetFetcher(offsetCriteria, partitionId, streamConfig);
    try {
      RetryPolicies.fixedDelayRetryPolicy(3, 1000L).attempt(partitionOffsetFetcher);
      return partitionOffsetFetcher.getOffset();
    } catch (Exception e) {
      Exception fetcherException = partitionOffsetFetcher.getException();
      LOGGER.error("Could not get offset for topic {} partition {}, criteria {}", streamConfig.getTopicName(),
          partitionId, offsetCriteria, fetcherException);
      throw new RuntimeException(fetcherException);
    }
  }

  private long getBetterStartOffsetIfNeeded(final String realtimeTableName, final int partition,
      final LLCSegmentName latestSegment, final long oldestOffsetInStream, final int nextSeqNum) {
    final LLCRealtimeSegmentZKMetadata oldSegMetadata =
        getRealtimeSegmentZKMetadata(realtimeTableName, latestSegment.getSegmentName(), null);
    CommonConstants.Segment.Realtime.Status status = oldSegMetadata.getStatus();
    long segmentStartOffset = oldestOffsetInStream;
    // Offset at which the prev segment intended to start consuming
    final long prevSegStartOffset = oldSegMetadata.getStartOffset();
    if (status.equals(CommonConstants.Segment.Realtime.Status.IN_PROGRESS)) {
      if (oldestOffsetInStream <= prevSegStartOffset) {
        // We still have the same start offset available, re-use it.
        segmentStartOffset = prevSegStartOffset;
        LOGGER.info("Choosing previous segment start offset {} for table {} for partition {}, sequence {}",
            oldestOffsetInStream, realtimeTableName, partition, nextSeqNum);
      } else {
        // There is data loss.
        LOGGER.warn("Data lost from offset {} to {} for table {} partition {} sequence {}", prevSegStartOffset,
            oldestOffsetInStream, realtimeTableName, partition, nextSeqNum);
        // Start from the earliest offset in the stream
        _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_KAFKA_DATA_LOSS, 1);
        _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_STREAM_DATA_LOSS, 1);
      }
    } else {
      // Status must be DONE, so we have a valid end-offset for the previous segment
      final long prevSegEndOffset = oldSegMetadata.getEndOffset();  // Will be 0 if the prev segment was not completed.
      if (oldestOffsetInStream < prevSegEndOffset) {
        // We don't want to create a segment that overlaps in data with the prev segment. We know that the previous
        // segment's end offset is available, so use that.
        segmentStartOffset = prevSegEndOffset;
        LOGGER.info("Choosing newer offset {} for table {} for partition {}, sequence {}", oldestOffsetInStream,
            realtimeTableName, partition, nextSeqNum);
      } else if (oldestOffsetInStream > prevSegEndOffset) {
        // Stream's oldest offset is greater than the end offset of the prev segment, so there is data loss.
        LOGGER.warn("Data lost from offset {} to {} for table {} partition {} sequence {}", prevSegEndOffset,
            oldestOffsetInStream, realtimeTableName, partition, nextSeqNum);
        _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_KAFKA_DATA_LOSS, 1);
        _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_STREAM_DATA_LOSS, 1);
      } else {
        // The two happen to be equal. A rarity, so log it.
        LOGGER.info("Earliest offset {} is the same as new segment start offset", oldestOffsetInStream);
      }
    }
    return segmentStartOffset;
  }

  private ZNRecord makeZnRecordForNewSegment(String realtimeTableName, int numReplicas, long startOffset,
      LLCSegmentName newSegmentName, int numPartitions) {
    final LLCRealtimeSegmentZKMetadata newSegMetadata = new LLCRealtimeSegmentZKMetadata();
    newSegMetadata.setCreationTime(System.currentTimeMillis());
    newSegMetadata.setStartOffset(startOffset);
    newSegMetadata.setEndOffset(END_OFFSET_FOR_CONSUMING_SEGMENTS);
    newSegMetadata.setNumReplicas(numReplicas);
    newSegMetadata.setTableName(realtimeTableName);
    newSegMetadata.setSegmentName(newSegmentName.getSegmentName());
    newSegMetadata.setStatus(CommonConstants.Segment.Realtime.Status.IN_PROGRESS);

    // Add the partition metadata if available.
    SegmentPartitionMetadata partitionMetadata =
        getPartitionMetadataFromTableConfig(realtimeTableName, numPartitions, newSegmentName.getPartitionId());
    if (partitionMetadata != null) {
      newSegMetadata.setPartitionMetadata(partitionMetadata);
    }

    return newSegMetadata.toZNRecord();
  }

  /**
   * An instance is reporting that it has stopped consuming a topic due to some error.
   * Mark the state of the segment to be OFFLINE in idealstate.
   * When all replicas of this segment are marked offline, the ValidationManager, in its next
   * run, will auto-create a new segment with the appropriate offset.
   */
  public void segmentStoppedConsuming(final LLCSegmentName segmentName, final String instance) {
    String rawTableName = segmentName.getTableName();
    String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
    final String segmentNameStr = segmentName.getSegmentName();
    try {
      HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
        idealState.setPartitionState(segmentNameStr, instance,
            CommonConstants.Helix.StateModel.SegmentOnlineOfflineStateModel.OFFLINE);
        Map<String, String> instanceStateMap = idealState.getInstanceStateMap(segmentNameStr);
        LOGGER.info("Attempting to mark {} offline. Current map:{}", segmentNameStr, instanceStateMap.toString());
        return idealState;
      }, RetryPolicies.exponentialBackoffRetryPolicy(10, 500L, 1.2f));
    } catch (Exception e) {
      LOGGER.error("Failed to update idealstate for table {} instance {} segment {}", realtimeTableName, instance,
          segmentNameStr, e);
      _controllerMetrics.addMeteredGlobalValue(ControllerMeter.LLC_ZOOKEEPER_UPDATE_FAILURES, 1);
      throw e;
    }
    LOGGER.info("Successfully marked {} offline for instance {} since it stopped consuming", segmentNameStr, instance);
  }

  /**
   * Helper method to trigger metadata event notifier
   * @param tableName a table name
   */
  private void notifyOnSegmentFlush(String tableName) {
    final MetadataEventNotifierFactory metadataEventNotifierFactory =
        MetadataEventNotifierFactory.loadFactory(_controllerConf.subset(METADATA_EVENT_NOTIFIER_PREFIX));
    final TableConfig tableConfig = getRealtimeTableConfig(tableName);
    metadataEventNotifierFactory.create().notifyOnSegmentFlush(tableConfig);
  }

  protected int getPartitionCount(StreamConfig streamConfig) {
    return PinotTableIdealStateBuilder.getPartitionCount(streamConfig);
  }

  /**
   * Returns the LLC realtime segments for the given table.
   *
   * @param realtimeTableName Realtime table name
   * @return List of LLC realtime segment names
   */
  @VisibleForTesting
  protected List<String> getAllSegments(String realtimeTableName) {
    return ZKMetadataProvider.getLLCRealtimeSegments(_propertyStore, realtimeTableName);
  }

  /**
   * Returns the LLC realtime segment ZK metadata for the given table and segment.
   *
   * @param realtimeTableName Realtime table name
   * @param segmentName Segment name (String)
   * @return LLC realtime segment ZK metadata
   */
  @VisibleForTesting
  protected LLCRealtimeSegmentZKMetadata getSegmentMetadata(String realtimeTableName, String segmentName) {
    return (LLCRealtimeSegmentZKMetadata) ZKMetadataProvider.getRealtimeSegmentZKMetadata(_propertyStore,
        realtimeTableName, segmentName);
  }

  /**
   * Returns the latest 2 LLC realtime segment ZK metadata for each partition.
   *
   * @param realtimeTableName Realtime table name
   * @return Map from partition to array of latest LLC realtime segment ZK metadata
   */
  @VisibleForTesting
  protected Map<Integer, LLCRealtimeSegmentZKMetadata[]> getLatestMetadata(String realtimeTableName) {
    List<String> llcRealtimeSegments = getAllSegments(realtimeTableName);

    Map<Integer, LLCSegmentName[]> partitionToLatestSegmentsMap = new HashMap<>();
    for (String llcRealtimeSegment : llcRealtimeSegments) {
      LLCSegmentName segmentName = new LLCSegmentName(llcRealtimeSegment);
      partitionToLatestSegmentsMap.compute(segmentName.getPartitionId(), (partitionId, segmentNames) -> {
        if (segmentNames == null) {
          return new LLCSegmentName[]{segmentName, null};
        } else {
          if (segmentName.getSequenceNumber() > segmentNames[0].getSequenceNumber()) {
            segmentNames[1] = segmentNames[0];
            segmentNames[0] = segmentName;
          } else if (segmentNames[1] == null || segmentName.getSequenceNumber() > segmentNames[1].getSequenceNumber()) {
            segmentNames[1] = segmentName;
          }
          return segmentNames;
        }
      });
    }

    Map<Integer, LLCRealtimeSegmentZKMetadata[]> partitionToLatestMetadataMap = new HashMap<>();
    for (Map.Entry<Integer, LLCSegmentName[]> entry : partitionToLatestSegmentsMap.entrySet()) {
      LLCSegmentName[] latestSegments = entry.getValue();
      LLCRealtimeSegmentZKMetadata latestMetadata =
          getSegmentMetadata(realtimeTableName, latestSegments[0].getSegmentName());
      LLCRealtimeSegmentZKMetadata secondLatestMetadata = null;
      if (latestSegments[1] != null) {
        secondLatestMetadata = getSegmentMetadata(realtimeTableName, latestSegments[1].getSegmentName());
      }
      partitionToLatestMetadataMap.put(entry.getKey(),
          new LLCRealtimeSegmentZKMetadata[]{latestMetadata, secondLatestMetadata});
    }

    return partitionToLatestMetadataMap;
  }

  /**
   * Validates llc segments in ideal state and repairs them if necessary
   *
   * During segment commit, we update zookeeper in 3 steps
   * Step 1: Update PROPERTYSTORE to change the old segment metadata status to DONE
   * Step 2: Update PROPERTYSTORE to create the new segment metadata with status IN_PROGRESS
   * Step 3: Update IDEALSTATES to include new segment in CONSUMING state, and change old segment to ONLINE state.
   *
   * The controller may fail between these three steps.
   * So when validation manager runs, it needs to check the following:
   *
   * If it fails between step-1 and step-2:
   * Check whether there are any segments in the PROPERTYSTORE with status DONE, but no new segment in status IN_PROGRESS,
   * and hence the status of the segment in the IDEALSTATE is still CONSUMING
   *
   * If it fails between step-2 and-3:
   * Check whether there are any segments in PROPERTYSTORE with status IN_PROGRESS, that are not accounted for in idealState.
   * If so, it should create the new segments in idealState.
   *
   * If the controller fails after step-3, we are fine because the idealState has the new segments.
   * If the controller fails before step-1, the server will see this as an upload failure, and will re-try.
   * @param tableConfig
   *
   * TODO: We need to find a place to detect and update a gauge for nonConsumingPartitionsCount for a table, and reset it to 0 at the end of validateLLC
   */
  public void validateLLCSegments(final TableConfig tableConfig) {
    final String tableNameWithType = tableConfig.getTableName();
    final StreamConfig streamConfig = new StreamConfig(tableConfig.getIndexingConfig().getStreamConfigs());
    final int partitionCount = getPartitionCount(streamConfig);
    HelixHelper.updateIdealState(_helixManager, tableNameWithType, new Function<IdealState, IdealState>() {
      @Nullable
      @Override
      public IdealState apply(@Nullable IdealState idealState) {
        return validateLLCSegments(tableConfig, idealState, partitionCount);
      }
    }, RetryPolicies.exponentialBackoffRetryPolicy(10, 1000L, 1.2f), true);
  }

  /**
   * Updates ideal state after completion of a realtime segment
   * @param tableNameWithType
   * @param currentSegmentId
   * @param newSegmentId
   * @param partitionAssignment
   */
  @VisibleForTesting
  protected void updateIdealStateOnSegmentCompletion(@Nonnull final String tableNameWithType,
      @Nonnull final String currentSegmentId, @Nonnull final String newSegmentId,
      @Nonnull final PartitionAssignment partitionAssignment) {

    HelixHelper.updateIdealState(_helixManager, tableNameWithType, new Function<IdealState, IdealState>() {
      @Nullable
      @Override
      public IdealState apply(@Nullable IdealState idealState) {
        return updateIdealStateOnSegmentCompletion(idealState, currentSegmentId, newSegmentId, partitionAssignment);
      }
    }, RetryPolicies.exponentialBackoffRetryPolicy(10, 1000L, 1.2f));
  }

  /**
   * Sets up a new table's segments metadata and returns the ideal state setup with initial segments
   * @param tableConfig
   * @param idealState
   * @param partitionCount
   * @return
   */
  private IdealState setupTable(TableConfig tableConfig, IdealState idealState, int partitionCount)
      throws InvalidConfigException {
    final String tableNameWithType = tableConfig.getTableName();
    if (!idealState.isEnabled()) {
      LOGGER.info("Skipping validation for disabled table {}", tableNameWithType);
      return idealState;
    }
    final StreamConfig streamConfig = new StreamConfig(tableConfig.getIndexingConfig().getStreamConfigs());
    final long now = getCurrentTimeMs();

    PartitionAssignment partitionAssignment =
        _streamPartitionAssignmentGenerator.generateStreamPartitionAssignment(tableConfig, partitionCount);

    Set<Integer> newPartitions = new HashSet<>(partitionCount);
    for (int partition = 0; partition < partitionCount; partition++) {
      newPartitions.add(partition);
    }

    OffsetCriteria offsetCriteria = streamConfig.getOffsetCriteria();
    Set<String> consumingSegments =
        setupNewPartitions(tableConfig, streamConfig, offsetCriteria, partitionAssignment, newPartitions, now);

    RealtimeSegmentAssignmentStrategy segmentAssignmentStrategy = new ConsumingSegmentAssignmentStrategy();
    Map<String, List<String>> assignments = segmentAssignmentStrategy.assign(consumingSegments, partitionAssignment);

    updateIdealState(idealState, null, consumingSegments, assignments);
    return idealState;
  }

  /*
   *  A segment commit takes 3 modifications to zookeeper:
   *  - Change old segment metadata (mark it DONE, and then other things)
   *  - Add new metadata
   *  - Update idealstate to change oldsegment to ONLINE and new segment to CONSUMING
   *
   * A controller failure before/during the first step, or after the last step, does not bother us.
   * However, a controller failure after the first step (but before the successful completion of last step)
   * will result in leaving zookeeper in an inconsistent state. We have logic to trigger a periodic scan
   * of the segments, and repair those in this intermediate stage of segment completion.
   *
   * Now that there are two threads that may try to create new segments, we need to be careful.
   *
   * It may happen that the segment completion thread has just done step-1, but meanwhile the periodic
   * validator triggers, and mistakes this segment to be in incomplete state.
   *
   * We check the segment's metadata to see if that is old enough for repair. If it is fairly new, we
   * leave it as it is, to be fixed the next time repair job triggers.
   */
  @VisibleForTesting
  protected boolean isTooSoonToCorrect(String tableNameWithType, String segmentId, long now) {
    Stat stat = new Stat();
    LLCRealtimeSegmentZKMetadata metadata = getRealtimeSegmentZKMetadata(tableNameWithType, segmentId, stat);
    long metadataUpdateTime = stat.getMtime();
    if (now < metadataUpdateTime + TimeUnit.MILLISECONDS.convert(MAX_SEGMENT_COMPLETION_TIME_MINS, TimeUnit.MINUTES)) {
      LOGGER.info("Too soon to correct segment:{} updateTime: {} now:{}", segmentId, metadataUpdateTime, now);
      return true;
    }
    return false;
  }

  private boolean isAllInstancesInState(Map<String, String> instanceStateMap, String state) {
    return instanceStateMap.values().stream().allMatch(value -> value.equals(state));
  }

  /*
   * Validate LLC segments of a table.
   *
   * Iterates over latest metadata for each partition and checks for following scenarios and repairs them:
   * 1) Segment present in ideal state
   * a) metadata status is IN_PROGRESS, segment state is CONSUMING - happy path
   * b) metadata status is IN_PROGRESS, segment state is OFFLINE - create new metadata and new CONSUMING segment
   * c) metadata status is DONE, segment state is OFFLINE - create new metadata and new CONSUMING segment
   * d) metadata status is DONE, segment state is CONSUMING - create new metadata and new CONSUMING segment
   * 2) Segment is absent from ideal state - add new segment to ideal state
   *
   * Also checks if it is too soon to correct (could be in the process of committing segment)
   * If new partitions are detected, gets them started
   * If new instances are detected, uses them for new assignments
   *
   * So, the method may:
   * - Add or modify one or more segment metadata znodes
   * - Update the idealstate, which may fail if some other process updated the idealstate.
   *
   * In case idealstate update fails, then we need to start over.
   * TODO: split this method into multiple smaller methods
   */
  @VisibleForTesting
  protected IdealState validateLLCSegments(final TableConfig tableConfig, IdealState idealState,
      final int partitionCount) {
    final String tableNameWithType = tableConfig.getTableName();
    final StreamConfig streamConfig = new StreamConfig(tableConfig.getIndexingConfig().getStreamConfigs());
    if (!idealState.isEnabled()) {
      LOGGER.info("Skipping validation for disabled table {}", tableNameWithType);
      return idealState;
    }
    final long now = getCurrentTimeMs();

    // Get the metadata for the latest 2 segments of each partition
    Map<Integer, LLCRealtimeSegmentZKMetadata[]> partitionToLatestMetadata = getLatestMetadata(tableNameWithType);

    // Find partitions for which there is no metadata at all. These are new partitions that we need to start consuming.
    Set<Integer> newPartitions = new HashSet<>(partitionCount);
    for (int partition = 0; partition < partitionCount; partition++) {
      if (!partitionToLatestMetadata.containsKey(partition)) {
        LOGGER.info("Found partition {} with no segments", partition);
        newPartitions.add(partition);
      }
    }

    PartitionAssignment partitionAssignment;
    boolean skipNewPartitions = false;
    try {
      partitionAssignment =
          _streamPartitionAssignmentGenerator.generateStreamPartitionAssignment(tableConfig, partitionCount);
    } catch (InvalidConfigException e) {
      _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.PARTITION_ASSIGNMENT_GENERATION_ERROR,
          1L);
      LOGGER.warn(
          "Could not generate partition assignment. Fetching partition assignment from ideal state for repair of table {}",
          tableNameWithType);
      partitionAssignment =
          _streamPartitionAssignmentGenerator.getStreamPartitionAssignmentFromIdealState(tableConfig, idealState);
      skipNewPartitions = true;
    }

    Set<String> onlineSegments = new HashSet<>(); // collect all segment names which should be updated to ONLINE state
    Set<String> consumingSegments =
        new HashSet<>(); // collect all segment names which should be created in CONSUMING state

    // Walk over all partitions that we have metadata for, and repair any partitions necessary.
    // Possible things to repair:
    // 1. The latest metadata is in DONE state, but the idealstate says segment is CONSUMING:
    //    a. Create metadata for next segment and find hosts to assign it to.
    //    b. update current segment in idealstate to ONLINE
    //    c. add new segment in idealstate to CONSUMING on the hosts.
    // 2. The latest metadata is IN_PROGRESS, but segment is not there in idealstate.
    //    a. change prev segment to ONLINE in idealstate
    //    b. add latest segment to CONSUMING in idealstate.
    // 3. All instances of a segment are in OFFLINE state.
    //    a. Create a new segment (with the next seq number)
    //       and restart consumption from the same offset (if possible) or a newer offset (if realtime stream does not have the same offset).
    //       In latter case, report data loss.
    for (Map.Entry<Integer, LLCRealtimeSegmentZKMetadata[]> entry : partitionToLatestMetadata.entrySet()) {
      int partition = entry.getKey();
      LLCRealtimeSegmentZKMetadata[] latestMetadataArray = entry.getValue();
      LLCRealtimeSegmentZKMetadata latestMetadata = latestMetadataArray[0];
      final String segmentId = latestMetadata.getSegmentName();
      final LLCSegmentName segmentName = new LLCSegmentName(segmentId);

      Map<String, Map<String, String>> mapFields = idealState.getRecord().getMapFields();
      if (mapFields.containsKey(segmentId)) {
        // Latest segment of metadata is in idealstate.
        Map<String, String> instanceStateMap = idealState.getInstanceStateMap(segmentId);
        if (instanceStateMap.values().contains(PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE)) {
          if (latestMetadata.getStatus().equals(CommonConstants.Segment.Realtime.Status.DONE)) {

            // step-1 of commmitSegmentMetadata is done (i.e. marking old segment as DONE)
            // but step-2 is not done (i.e. adding new metadata for the next segment)
            // and ideal state update (i.e. marking old segment as ONLINE and new segment as CONSUMING) is not done either.
            if (isTooSoonToCorrect(tableNameWithType, segmentId, now)) {
              continue;
            }
            LOGGER.info("{}:Repairing segment for partition {}. "
                    + "Old segment metadata {} has status DONE, but segments are still in CONSUMING state in ideal STATE",
                tableNameWithType, partition, segmentId);

            LLCSegmentName newLLCSegmentName = makeNextLLCSegmentName(segmentName, partition, now);
            LOGGER.info("{}: Creating new segment metadata for {}", tableNameWithType,
                newLLCSegmentName.getSegmentName());

            CommittingSegmentDescriptor committingSegmentDescriptor =
                new CommittingSegmentDescriptor(segmentId, latestMetadata.getEndOffset(), 0);
            boolean success =
                createNewSegmentMetadataZNRecord(tableConfig, segmentName, newLLCSegmentName, partitionAssignment,
                    committingSegmentDescriptor, false);

            // creation of segment metadata could fail due to lost leadership or an unsuccessful write to property store
            // in such a case, we will exclude the segment from ideal state update and let the next iteration of validation manager fix it
            if (success) {
              onlineSegments.add(segmentId);
              consumingSegments.add(newLLCSegmentName.getSegmentName());
            }
          }
          // else, the metadata should be IN_PROGRESS, which is the right state for a consuming segment.
        } else { // no replica in CONSUMING state

          // Possible scenarios: for any of these scenarios, we need to create new metadata IN_PROGRESS and new CONSUMING segment
          // 1. all replicas OFFLINE and metadata IN_PROGRESS/DONE - a segment marked itself OFFLINE during consumption for some reason
          // 2. all replicas ONLINE and metadata DONE - Resolved in https://github.com/linkedin/pinot/pull/2890
          // 3. we should never end up with some replicas ONLINE and some OFFLINE.
          if (isAllInstancesInState(instanceStateMap, PinotHelixSegmentOnlineOfflineStateModelGenerator.OFFLINE_STATE)
              || !(isTooSoonToCorrect(tableNameWithType, segmentId, now))) {

            // No instances are consuming, so create a new consuming segment.
            LLCSegmentName newLLCSegmentName = makeNextLLCSegmentName(segmentName, partition, now);
            LOGGER.info("Creating CONSUMING segment {} for {} partition {}", newLLCSegmentName.getSegmentName(),
                tableNameWithType, partition);

            // To begin with, set startOffset to the oldest available offset in the stream. Fix it to be the one we want,
            // depending on what the prev segment had.
            long startOffset = getPartitionOffset(streamConfig, OffsetCriteria.SMALLEST_OFFSET_CRITERIA, partition);
            LOGGER.info("Found smallest offset {} for table {} for partition {}", startOffset, tableNameWithType,
                partition);
            startOffset = getBetterStartOffsetIfNeeded(tableNameWithType, partition, segmentName, startOffset,
                newLLCSegmentName.getSequenceNumber());

            CommittingSegmentDescriptor committingSegmentDescriptor =
                new CommittingSegmentDescriptor(segmentId, startOffset, 0);

            boolean success =
                createNewSegmentMetadataZNRecord(tableConfig, segmentName, newLLCSegmentName, partitionAssignment,
                    committingSegmentDescriptor, false);

            // creation of segment metadata could fail due to lost leadership or an unsuccessful write to property store
            // in such a case, we will exclude the segment from ideal state update and let the next iteration of validation manager fix it
            if (success) {
              consumingSegments.add(newLLCSegmentName.getSegmentName());
            }
          }
        }
      } else {
        // idealstate does not have an entry for the segment (but metadata is present)
        // controller has failed between step-2 and step-3 of commitSegmentMetadata.
        // i.e. after updating old segment metadata (old segment metadata state = DONE)
        // and creating new segment metadata (new segment metadata state = IN_PROGRESS),
        // but before updating ideal state (new segment ideal missing from ideal state)
        if (isTooSoonToCorrect(tableNameWithType, segmentId, now)) {
          continue;
        }

        Preconditions.checkArgument(
            latestMetadata.getStatus().equals(CommonConstants.Segment.Realtime.Status.IN_PROGRESS));
        LOGGER.info("{}:Repairing segment for partition {}. Segment {} not found in idealstate", tableNameWithType,
            partition, segmentId);

        // If there was a prev segment in the same partition, then we need to fix it to be ONLINE.
        LLCRealtimeSegmentZKMetadata secondLatestMetadata = latestMetadataArray[1];
        if (secondLatestMetadata == null && skipNewPartitions) {
          continue;
        }
        if (secondLatestMetadata != null) {
          onlineSegments.add(secondLatestMetadata.getSegmentName());
        }
        consumingSegments.add(segmentId);
      }
    }

    if (!skipNewPartitions) {
      Set<String> newPartitionSegments =
          setupNewPartitions(tableConfig, streamConfig, OffsetCriteria.SMALLEST_OFFSET_CRITERIA, partitionAssignment,
              newPartitions, now);
      consumingSegments.addAll(newPartitionSegments);
    }

    RealtimeSegmentAssignmentStrategy segmentAssignmentStrategy = new ConsumingSegmentAssignmentStrategy();
    Map<String, List<String>> assignments;
    try {
      assignments = segmentAssignmentStrategy.assign(consumingSegments, partitionAssignment);
    } catch (InvalidConfigException e) {
      throw new IllegalStateException(
          "Caught exception when assigning segments using partition assignment for table " + tableNameWithType);
    }

    updateIdealState(idealState, onlineSegments, consumingSegments, assignments);
    return idealState;
  }

  private LLCSegmentName makeNextLLCSegmentName(LLCSegmentName segmentName, int partition, long now) {
    final int newSeqNum = segmentName.getSequenceNumber() + 1;
    LLCSegmentName newLLCSegmentName = new LLCSegmentName(segmentName.getTableName(), partition, newSeqNum, now);
    return newLLCSegmentName;
  }

  /**
   * Updates the ideal state object
   * Adds the segments in consumingSegments to CONSUMING state using instances from assignments
   * Sets the segments in onlineSegments to ONLINE state
   * @param idealState
   * @param consumingSegments
   * @param onlineSegments
   * @param assignments
   */
  private void updateIdealState(IdealState idealState, Set<String> onlineSegments, Set<String> consumingSegments,
      Map<String, List<String>> assignments) {
    if (onlineSegments != null) {
      for (String segment : onlineSegments) {
        Set<String> oldInstances = idealState.getInstanceSet(segment);
        Preconditions.checkArgument(CollectionUtils.isNotEmpty(oldInstances));
        for (String instance : oldInstances) {
          idealState.setPartitionState(segment, instance,
              PinotHelixSegmentOnlineOfflineStateModelGenerator.ONLINE_STATE);
        }
      }
    }

    if (consumingSegments != null) {
      for (String segment : consumingSegments) {
        List<String> newInstances = assignments.get(segment);
        Map<String, String> instanceStateMap = idealState.getInstanceStateMap(segment);
        if (instanceStateMap != null) {
          instanceStateMap.clear();
        }
        for (String instance : newInstances) {
          idealState.setPartitionState(segment, instance,
              PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE);
        }
      }
    }
  }

  /**
   * Create metadata for new partitions
   * @param tableConfig  the table configuration to use for the new partition
   * @param streamConfig stream configuration associated with the table
   * @param offsetCriteria the offset to query to start consumption from. Can be different for a
   *                       new table being setup vs new partitions being added to an existing table
   * @param partitionAssignment the partition assignment strategy to use
   * @param newPartitions the new partitions to set up
   * @param now the current timestamp in milliseconds
   * @return set of newly created segment names
   */
  private Set<String> setupNewPartitions(TableConfig tableConfig, StreamConfig streamConfig,
      OffsetCriteria offsetCriteria, PartitionAssignment partitionAssignment, Set<Integer> newPartitions, long now) {

    String tableName = tableConfig.getTableName();
    Set<String> newSegmentNames = new HashSet<>(newPartitions.size());
    String rawTableName = TableNameBuilder.extractRawTableName(tableName);
    int nextSeqNum = STARTING_SEQUENCE_NUMBER;

    for (int partition : newPartitions) {
      LOGGER.info("Creating CONSUMING segment for {} partition {} with seq {}", tableName, partition,
          nextSeqNum);
      long startOffset = getPartitionOffset(streamConfig, offsetCriteria, partition);

      LOGGER.info("Found offset {} for table {} for partition {}", startOffset, tableName, partition);

      LLCSegmentName newLLCSegmentName = new LLCSegmentName(rawTableName, partition, nextSeqNum, now);
      CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(null, startOffset, 0);

      boolean success = createNewSegmentMetadataZNRecord(tableConfig, null, newLLCSegmentName, partitionAssignment,
          committingSegmentDescriptor, true);
      // creation of segment metadata could fail due to an unsuccessful write to property store
      // in such a case, we will exclude the segment from ideal state update and let the validation manager fix it
      if (success) {
        newSegmentNames.add(newLLCSegmentName.getSegmentName());
      }
    }
    return newSegmentNames;
  }

  @VisibleForTesting
  protected long getCurrentTimeMs() {
    return System.currentTimeMillis();
  }

  protected IdealState updateIdealStateOnSegmentCompletion(@Nonnull IdealState idealState,
      @Nonnull String currentSegmentId, @Nonnull String newSegmentId,
      @Nonnull PartitionAssignment partitionAssignment) {

    Map<String, List<String>> instanceAssignments = null;

    RealtimeSegmentAssignmentStrategy strategy = new ConsumingSegmentAssignmentStrategy();
    try {
      instanceAssignments = strategy.assign(Lists.newArrayList(newSegmentId), partitionAssignment);
    } catch (InvalidConfigException e) {
      _controllerMetrics.addMeteredTableValue(idealState.getResourceName(),
          ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
      throw new IllegalStateException("Caught exception when updating ideal state on segment completion", e);
    }

    List<String> newSegmentInstances = instanceAssignments.get(newSegmentId);
    Set<String> currentSegmentInstances = idealState.getInstanceSet(currentSegmentId);
    for (String instance : currentSegmentInstances) {
      idealState.setPartitionState(currentSegmentId, instance,
          PinotHelixSegmentOnlineOfflineStateModelGenerator.ONLINE_STATE);
    }

    // We may have (for whatever reason) a different instance list in the idealstate for the new segment.
    // If so, clear it, and then set the instance state for the set of instances that we know should be there.
    Map<String, String> stateMap = idealState.getInstanceStateMap(newSegmentId);
    if (stateMap != null) {
      stateMap.clear();
    }
    for (String instance : newSegmentInstances) {
      idealState.setPartitionState(newSegmentId, instance,
          PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE);
    }

    return idealState;
  }
}


text:
=head1 LICENSE

  Copyright (c) 1999-2011 The European Bioinformatics Institute and
  Genome Research Limited.  All rights reserved.

  This software is distributed under a modified Apache license.
  For license details, please see

    http://www.ensembl.org/info/about/code_licence.html

=head1 CONTACT

  Please email comments or questions to the public Ensembl
  developers list at <dev@ensembl.org>.

  Questions may also be sent to the Ensembl help desk at
  <helpdesk@ensembl.org>.

=cut

=head1 NAME

Bio::EnsEMBL::RepeatMaskedSlice - Arbitary Slice of a genome

=head1 SYNOPSIS

  $sa = $db->get_SliceAdaptor();

  $slice =
    $sa->fetch_by_region( 'chromosome', 'X', 1_000_000, 2_000_000 );

  $repeat_masked_slice = $slice->get_repeatmasked_seq();

  # get repeat masked sequence:
  my $dna = $repeat_masked_slice->seq();
  $dna = $repeat_masked_slice->subseq( 1, 1000 );

=head1 DESCRIPTION

This is a specialised Bio::EnsEMBL::Slice class that is used to retrieve
repeat masked genomic sequence rather than normal genomic sequence.

=head1 METHODS

=cut

package Bio::EnsEMBL::RepeatMaskedSlice;

use strict;
use warnings;

use Bio::EnsEMBL::Slice;
use Bio::EnsEMBL::Utils::Argument qw(rearrange);
use Bio::EnsEMBL::Utils::Sequence qw(reverse_comp);
use Bio::EnsEMBL::Utils::Exception;

use vars qw(@ISA);

@ISA = ('Bio::EnsEMBL::Slice');

# The BLOCK_PWR is the lob_bin of the chunksize where you want your repeat features
# to be retreived. This will create repeat feature retrieval calls that are likely
# to be on the same slice and hopefully create cache hits and less database traffic
my $BLOCK_PWR = 18;

=head2 new

  Arg [-REPEAT_MASK] : The logic name of the repeats to be used for masking.
                      If not provided, all repeats in the database are used.
  Arg [...]  : Named superclass arguments. See B<Bio::EnsEMBL::Slice>.
  Example    : my $slice = Bio::EnsEMBL::RepeatMaskedSlice->new
                  (-START  => $start,
                   -END    => $end,
                   -STRAND => $strand,
                   -SEQ_REGION_NAME => $seq_region,
                   -SEQ_REGION_LENGTH => $seq_region_length,
                   -COORD_SYSTEM  => $cs,
                   -ADAPTOR => $adaptor,
                   -REPEAT_MASK => ['repeat_masker'],
                   -SOFT_MASK => 1,
                   -NOT_DEFAULT_MASKING_CASES => {"repeat_class_SINE/MIR" => 1,
                                                  "repeat_name_AluSp" => 0});
  Description: Creates a Slice which behaves exactly as a normal slice but
               that returns repeat masked sequence from the seq method.
  Returntype : Bio::EnsEMBL::RepeatMaskedSlice
  Exceptions : none
  Caller     : RawComputes (PredictionTranscript creation code).
  Status     : Stable

=cut

sub new {
  my $caller = shift;
  my $class = ref($caller) || $caller;

  my ($logic_names, $soft_mask, $not_default_masking_cases) = rearrange(['REPEAT_MASK',
                                                                         'SOFT_MASK',
                                                                         'NOT_DEFAULT_MASKING_CASES'], @_);

  my $self = $class->SUPER::new(@_);

  $logic_names ||= [''];
  if(ref($logic_names) ne 'ARRAY') {
    throw("Reference to list of logic names argument expected.");
  }

  $self->{'repeat_mask_logic_names'} = $logic_names;
  $self->{'soft_mask'} = $soft_mask;
  $self->{'not_default_masking_cases'} = $not_default_masking_cases;
  $self->{'not_default_masking_cases'} ||= {};

  return $self;
}

=head2 repeat_mask_logic_names

  Arg [1]    : reference to list of strings $logic_names (optional)
  Example    : $rm_slice->repeat_mask_logic_name(['repeat_masker']);
  Description: Getter/Setter for the logic_names of the repeats that are used
               to mask this slices sequence.
  Returntype : reference to list of strings
  Exceptions : none
  Caller     : seq() method
  Status     : Stable

=cut

sub repeat_mask_logic_names {
  my $self = shift;

  if(@_) {
    my $array = shift;
    if(ref($array) ne 'ARRAY') {
      throw('Reference to list of logic names argument expected.');
    }
  }

  return $self->{'repeat_mask_logic_names'};
}

=head2 soft_mask

  Arg [1]    : boolean $soft_mask (optional)
  Example    : $rm_slice->soft_mask(0);
  Description: Getter/Setter which is used to turn on/off softmasking of the
               sequence returned by seq.
  Returntype : boolean
  Exceptions : none
  Caller     : seq() method
  Status     : Stable

=cut

sub soft_mask {
  my $self = shift;
  $self->{'soft_mask'} = shift if(@_);
  return $self->{'soft_mask'} || 0;
}

=head2 not_default_masking_cases

  Arg [1]    : hash reference $not_default_masking_cases (optional, default is {})
               The values are 0 or 1 for hard and soft masking respectively
               The keys of the hash should be of 2 forms
               "repeat_class_" . $repeat_consensus->repeat_class,
                e.g. "repeat_class_SINE/MIR"
               "repeat_name_" . $repeat_consensus->name
                e.g. "repeat_name_MIR"
               depending on which base you want to apply the not default masking either
               the repeat_class or repeat_name. Both can be specified in the same hash
               at the same time, but in that case, repeat_name setting has priority over
               repeat_class. For example, you may have hard masking as default, and
               you may want soft masking of all repeat_class SINE/MIR,
               but repeat_name AluSp (which are also from repeat_class SINE/MIR)
  Example    : $rm_slice->not_default_masking_cases({"repeat_class_SINE/MIR" => 1,
                                                     "repeat_name_AluSp" => 0});
  Description: Getter/Setter which is used to escape some repeat class or name from the default
               masking in place.
  Returntype : hash reference
  Exceptions : none
  Caller     : seq() and subseq() methods
  Status     : Stable

=cut

sub not_default_masking_cases {
  my $self = shift;
  $self->{'not_default_masking_cases'} = shift if (@_);
  return $self->{'not_default_masking_cases'};
}

=head2 seq

  Arg [1]    : none
  Example    : print $rmslice->seq(), "\n";
  Description: Retrieves the entire repeat masked sequence for this slice.
               See also the B<Bio::EnsEMBL::Slice> implementation of this
               method.
  Returntype : string
  Exceptions : none
  Caller     : general
  Status     : Stable

=cut

sub seq {
  my $self = shift;
  #
  # get all the features
  #
  my $logic_names = $self->repeat_mask_logic_names();
  my $soft_mask   = $self->soft_mask();
  my $not_default_masking_cases = $self->not_default_masking_cases;

  my $repeats = [];

  foreach my $l (@$logic_names) {
    push @{$repeats}, @{$self->get_all_RepeatFeatures($l)};
  }

  #
  # get the dna
  #
  my $dna = $self->SUPER::seq(@_);

  #
  # mask the dna
  #
  $self->_mask_features(\$dna,$repeats,$soft_mask,$not_default_masking_cases);
  return $dna;
}

=head2 subseq

  Arg [1]    : none
  Example    : print $rmslice->subseq(1, 1000);
  Description: Retrieves a repeat masked sequence from a specified subregion
               of this slice.  See also the B<Bio::EnsEMBL::Slice>
               implementation of this method.
  Returntype : string
  Exceptions : none
  Caller     : general
  Status     : Stable

=cut

sub subseq {
  my $self   = shift;
  my $start  = shift;
  my $end    = shift;
  my $strand = shift;

  #
  # get all the features
  #
  my $logic_names = $self->repeat_mask_logic_names();
  my $soft_mask   = $self->soft_mask();
  my $not_default_masking_cases = $self->not_default_masking_cases;

  # If frequent subseqs happen on repeatMasked sequence this results in
  # a lot of feature retrieval from the database. To avoid this, features
  # are only retrieved from subslices with fixed space boundaries.
  # The access happens in block to make cache hits more likely

  # The blocksize can be defined on the top of this module.

  my $seq_region_slice = $self->seq_region_Slice();
  my $block_min = ($self->start()-1) >> $BLOCK_PWR;
  my $block_max = ($self->end()-1) >> $BLOCK_PWR;

  my $repeats = [];

  my $sub_start = ($block_min << $BLOCK_PWR)+1;
  my $sub_end = ($block_max+1)<<$BLOCK_PWR;
  if ($sub_end > $seq_region_slice->length) {
    $sub_end =  $seq_region_slice->length ;
  }

  my $subslice = $seq_region_slice->sub_Slice( $sub_start, $sub_end);

  foreach my $l (@$logic_names) {
    push @{$repeats}, @{$subslice->get_all_RepeatFeatures($l)};
  }

  #
  # get the dna
  #
  my $subsequence_slice = $self->sub_Slice( $start, $end, $strand );
  my $dna = $subsequence_slice->seq();
  #
  # mask the dna
  #
  $subsequence_slice->_mask_features(\$dna,$repeats,$soft_mask,$not_default_masking_cases);

  return $dna;
}

1;


text:
package org.feathersjs.feathersdemo;

import android.app.Activity;
import android.content.DialogInterface;
import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AlertDialog;
import android.util.Log;
import android.widget.EditText;

import org.feathersjs.client.Feathers;
import org.feathersjs.client.plugins.authentication.AuthResponse;
import org.feathersjs.client.service.FeathersService;
import org.feathersjs.feathersdemo.models.User;
import org.json.JSONObject;

import butterknife.Bind;
import butterknife.ButterKnife;
import butterknife.OnClick;

public class LoginActivity extends Activity {

    @Bind(R.id.username)
    EditText mUsernameEditText;

    @Bind(R.id.password)
    EditText mPasswordEditText;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_login);
        ButterKnife.bind(this);
        mUsernameEditText.setText("testing123@123.com");
        mPasswordEditText.setText("password");
    }

    protected void onDestroy() {
        super.onDestroy();
    }

    @OnClick(R.id.loginButton)
    void login() {

        String username = mUsernameEditText.getText().toString();
        String password = mPasswordEditText.getText().toString();

        Feathers.getInstance().authenticate(username, password, new FeathersService.FeathersCallback<AuthResponse<User>>() {
            @Override
            public void onSuccess(AuthResponse<User> t) {
                Intent intent = new Intent(LoginActivity.this, MainActivity.class);
                startActivity(intent);
            }

            @Override
            public void onError(String errorMessage) {
                Log.d("LoginActivity", "onError | " + errorMessage);

                AlertDialog.Builder builder1 = new AlertDialog.Builder(LoginActivity.this);
                builder1.setTitle("Error");
                builder1.setMessage(errorMessage);
                builder1.setCancelable(true);

                builder1.setPositiveButton(
                        "Yes",
                        new DialogInterface.OnClickListener() {
                            public void onClick(DialogInterface dialog, int id) {
                                dialog.cancel();
                            }
                        });

                builder1.setNegativeButton(
                        "No",
                        new DialogInterface.OnClickListener() {
                            public void onClick(DialogInterface dialog, int id) {
                                dialog.cancel();
                            }
                        });

                AlertDialog alert11 = builder1.create();
                alert11.show();
            }
        }, User.class);
    }
}