co.cask.cdap.data.stream

Class StreamUtils

java.lang.Object

co.cask.cdap.data.stream.StreamUtils

public final class StreamUtils
extends Object

Collection of helper methods. TODO: Usage of this class needs to be refactor, as some methods are temporary (e.g. encodeMap/decodeMap).

Method Summary

Methods

Modifier and Type	Method and Description
static Location	createGenerationLocation(Location streamBaseLocation, int generation) Creates stream base location with the given generation.
static Location	createPartitionLocation(Location baseLocation, long partitionStart, long partitionDuration) Creates the location for the partition directory.
static Location	createStreamLocation(Location partitionLocation, String prefix, int seqId, StreamFileType type) Creates location for stream file.
static StreamFileOffset	decodeOffset(StreamConfig config, DataInput in) Decode a StreamFileOffset encoded by the encodeOffset(DataOutput, StreamFileOffset) method.
static void	encodeOffset(DataOutput out, StreamFileOffset offset) Encode a StreamFileOffset instance.
static StreamConfig	ensureExists(StreamAdmin admin, Id.Stream streamId)
static long	fetchStreamFilesSize(Location streamLocation) Get the size of the data persisted for the stream under the given stream location.
static String	getBucketName(String name) Returns the name of the event bucket based on the file name.
static Location	getDeletedLocation(Location streamRootLocation) Returns the location of the stream deleted location.
static int	getGeneration(Location streamLocation) Finds the current generation if of a stream.
static int	getGeneration(StreamConfig config) Finds the current generation id of a stream.
static String	getNamePrefix(String name) Returns the file prefix based on the given file name.
static int	getNextSequenceId(Location partitionLocation, String filePrefix) Finds the next sequence id for the given partition with the given file prefix.
static long	getPartitionEndTime(String partitionName) Gets the partition end time based on the name of the partition.
static String	getPartitionName(Location eventLocation) Finds the partition name from the given event file location.
static long	getPartitionStartTime(long timestamp, long partitionDuration) Returns the aligned partition start time.
static long	getPartitionStartTime(String partitionName) Gets the partition start time based on the name of the partition.
static int	getSequenceId(String name) Returns the sequence number of the given file name.
static TableId	getStateStoreTableId(Id.Namespace namespace) Gets a TableId for stream consumer state stores within a given namespace.
static Id.Stream	getStreamIdFromLocation(Location streamBaseLocation) Gets a Id.Stream given a stream's base directory.
static boolean	isPartition(String partitionName) Returns true if it is valid partition name, false other.
static Iterable<Location>	listAllStreams(Location streamRootLocation) Lists all stream locations under the given root.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

getPartitionName

public static String getPartitionName(Location eventLocation)

Finds the partition name from the given event file location.

Parameters:

eventLocation - Location to the event file.

Returns:

The partition name.

See Also:

StreamInputFormat

getBucketName

public static String getBucketName(String name)

Returns the name of the event bucket based on the file name.

Parameters:

name - Name of the file.

See Also:

StreamInputFormat

getNamePrefix

public static String getNamePrefix(String name)

Returns the file prefix based on the given file name.

Parameters:

name - Name of the file.

Returns:

The prefix part of the stream file.

See Also:

StreamInputFormat

getSequenceId

public static int getSequenceId(String name)

Returns the sequence number of the given file name.

Parameters:

name - Name of the file.

Returns:

The sequence number of the stream file.

See Also:

StreamInputFormat

getPartitionStartTime

public static long getPartitionStartTime(String partitionName)

Gets the partition start time based on the name of the partition.

Returns:

The partition start timestamp in milliseconds.

See Also:

StreamInputFormat

isPartition

public static boolean isPartition(String partitionName)

Returns true if it is valid partition name, false other. The partition name must be [0-9]+.[0-9]+

getPartitionEndTime

public static long getPartitionEndTime(String partitionName)

Gets the partition end time based on the name of the partition.

Returns:

the partition end timestamp in milliseconds.

See Also:

StreamInputFormat

createGenerationLocation

public static Location createGenerationLocation(Location streamBaseLocation,
                                int generation)
                                         throws IOException

Creates stream base location with the given generation.

Parameters:

streamBaseLocation - the base directory for the stream

generation - generation id

Returns:

Location for the given generation

Throws:

IOException

See Also:

StreamInputFormat

createPartitionLocation

public static Location createPartitionLocation(Location baseLocation,
                               long partitionStart,
                               long partitionDuration)
                                        throws IOException

Creates the location for the partition directory.

Parameters:

baseLocation - Base location for partition directory.

partitionStart - Partition start timestamp in milliseconds.

partitionDuration - Partition duration in milliseconds.

Returns:

The location for the partition directory.

Throws:

IOException

createStreamLocation

public static Location createStreamLocation(Location partitionLocation,
                            String prefix,
                            int seqId,
                            StreamFileType type)
                                     throws IOException

Creates location for stream file.

Parameters:

partitionLocation - The partition directory location.

prefix - File prefix.

seqId - Sequence number of the file.

type - Type of the stream file.

Returns:

The location of the stream file.

Throws:

IOException

See Also:

for naming convention.

getPartitionStartTime

public static long getPartitionStartTime(long timestamp,
                         long partitionDuration)

Returns the aligned partition start time.

Parameters:

timestamp - Timestamp in milliseconds.

partitionDuration - Partition duration in milliseconds.

Returns:

The partition start time of the given timestamp.

encodeOffset

public static void encodeOffset(DataOutput out,
                StreamFileOffset offset)
                         throws IOException

Encode a StreamFileOffset instance.

Parameters:

out - Output for encoding

offset - The offset object to encode

Throws:

IOException

decodeOffset

public static StreamFileOffset decodeOffset(StreamConfig config,
                            DataInput in)
                                     throws IOException

Decode a StreamFileOffset encoded by the encodeOffset(DataOutput, StreamFileOffset) method.

Parameters:

config - Stream configuration for the stream that the offset is representing

in - Input for decoding

Returns:

A new instance of StreamFileOffset

Throws:

IOException

ensureExists

public static StreamConfig ensureExists(StreamAdmin admin,
                        Id.Stream streamId)
                                 throws IOException

Throws:

IOException

getGeneration

public static int getGeneration(StreamConfig config)
                         throws IOException

Finds the current generation id of a stream. It scans the stream directory to look for largest generation number in directory name.

Parameters:

config - configuration of the stream

Returns:

the generation id

Throws:

IOException

getGeneration

public static int getGeneration(Location streamLocation)
                         throws IOException

Finds the current generation if of a stream. It scans the stream directory to look for largest generation number in directory name.

Parameters:

streamLocation - location to scan for generation id

Returns:

the generation id

Throws:

IOException

getNextSequenceId

public static int getNextSequenceId(Location partitionLocation,
                    String filePrefix)
                             throws IOException

Finds the next sequence id for the given partition with the given file prefix.

Parameters:

partitionLocation - the directory where the stream partition is

filePrefix - prefix of file name to match

Returns:

the next sequence id, which is the current max id + 1.

Throws:

IOException - if failed to find the next sequence id

fetchStreamFilesSize

public static long fetchStreamFilesSize(Location streamLocation)
                                 throws IOException

Get the size of the data persisted for the stream under the given stream location.

Parameters:

streamLocation - stream to get data size of

Returns:

the size of the data persisted for the stream which config is the streamName

Throws:

IOException - in case of any error in fetching the size

getStateStoreTableId

public static TableId getStateStoreTableId(Id.Namespace namespace)

Gets a TableId for stream consumer state stores within a given namespace.

Parameters:

namespace - the namespace for which the table is for.

Returns:

constructed TableId

getStreamIdFromLocation

public static Id.Stream getStreamIdFromLocation(Location streamBaseLocation)

Gets a Id.Stream given a stream's base directory.

Parameters:

streamBaseLocation - the location of the stream's directory

Returns:

Id of the stream associated with the location

getDeletedLocation

public static Location getDeletedLocation(Location streamRootLocation)
                                   throws IOException

Returns the location of the stream deleted location.

Parameters:

streamRootLocation - the root location that all streams go under

Throws:

IOException

listAllStreams

public static Iterable<Location> listAllStreams(Location streamRootLocation)
                                         throws IOException

Lists all stream locations under the given root.

Parameters:

streamRootLocation - the root location that all streams go under

Throws:

IOException