Hdfs datasource tasker implementation #91

Tiihott · 2024-09-20T12:07:39Z

HDFS datasource feature to allow for querying the semi latest data from HDFS. Latest records that are not yet present in HDFS are queried from Kafka, old data is queried from S3 as usual.

The included planner and scheduler components were already reviewed outside GitHub.

Includes:

Migration of the existing hdfs datasource dev branch to GitHub, which included the planner and scheduler component implementations.
New tasker component implementation which reads records from avro-files that are stored in the HDFS.
Time based record inclusion for handling the cutoff between S3 and HDFS.
Kerberized HDFS access configurations.
New HdfsQueryProcessor.incrementAndGetLatestOffset() method for going through the offsets incrementally to split the data into appropriate sized batches.

Missing:

additional filtering of HDFS records based on queryXML which couldn't be implemented in the previously implemented planner and scheduler components.

src/main/java/com/teragrep/pth_06/ArchiveMicroStreamReader.java

src/main/java/com/teragrep/pth_06/FileSystemFactory.java

src/main/java/com/teragrep/pth_06/FileSystemFactoryImpl.java

src/main/java/com/teragrep/pth_06/HdfsTopicPartitionOffsetMetadata.java

src/main/java/com/teragrep/pth_06/config/HdfsConfig.java

src/main/java/com/teragrep/pth_06/planner/HdfsDBClient.java

src/main/java/com/teragrep/pth_06/planner/offset/DatasourceOffset.java

src/main/java/com/teragrep/pth_06/planner/offset/HdfsOffset.java

src/main/java/com/teragrep/pth_06/planner/offset/SerializedDatasourceOffset.java

src/main/java/com/teragrep/pth_06/scheduler/HdfsBatchSliceCollection.java

src/main/java/com/teragrep/pth_06/scheduler/Batch.java

src/main/java/com/teragrep/pth_06/task/hdfs/AvroReadImpl.java

src/main/java/com/teragrep/pth_06/task/hdfs/AvroRead.java

src/main/java/com/teragrep/pth_06/task/hdfs/HdfsRecordConversion.java

src/main/java/com/teragrep/pth_06/task/HdfsMicroBatchInputPartitionReader.java

src/main/java/com/teragrep/pth_06/task/hdfs/HdfsRecordConversionImpl.java

src/main/java/com/teragrep/pth_06/config/HdfsConfig.java

src/main/java/com/teragrep/pth_06/planner/HdfsDBClient.java

src/main/java/com/teragrep/pth_06/planner/HdfsQueryProcessor.java

src/main/java/com/teragrep/pth_06/planner/offset/DatasourceOffset.java

eemhu · 2024-10-25T10:57:29Z

src/main/java/com/teragrep/pth_06/planner/offset/KafkaOffset.java

        serializedKafkaOffset = new HashMap<>(offset.size());
        for (Map.Entry<TopicPartition, Long> entry : offset.entrySet()) {

            serializedKafkaOffset.put(entry.getKey().toString(), entry.getValue()); // offset
        }
+        this.stub = stub;
    }

    public KafkaOffset(String s) {
        Gson gson = new Gson();


could this be moved up and use primary constructor with this?

Map<TopicPartition, Long> is not serializable which is why it is converted to serializable Map<String, Long> in the constructor. And because of this logic being inside the constructors the KafkaOffset(String s) which is used by serialization testing can't be simply refactored to a secondary constructor.
The code must be refactored so the conversion to serializable map is done before the object is initialized, that way the constructor can be refactored into a secondary constructor.
Solved identical issue of HdfsOffset in commit: b928695
Solved in commit: cc07931

Could this potentially use a decorator object or similar?

The class extends the abstract class Offset and implements the interface Serializable. AFAIK, because the class is serializable it can't encapsulate the Map<TopicPartition, Long> and the same goes for decorators.

It would be possible to use transient keyword with the Map<TopicPartition, Long> which makes serialization ignore the class field. This way Map<TopicPartition, Long> could be used as input parameter for the constructor, and a new method can be added for converting the Map<TopicPartition, Long> into a Map<String, Long> before the serialization is done. That way at least the code duplication for converting the map to serializable format would be solved.
Implemented serializable Map conversion to HdfsOffset and KafkaOffset in commit 014cc64

…d properties.

…a and HdfsRecordConversionImpl.java to make all objects immutable. Time based inclusion refactoring is WIP.

…erializedDatasourceOffset.java to reduce the amount of if-statements.

…fset.java to remove null usages. Fixed error in initialOffset() method.

…uctor.

…empty BatchSliceCollection. Replaced initialization of BatchSliceCollection as null with StubBatchSliceCollection.

… inclusion functionality.

…tead of separate clear() method. Removed clear() from AvroRead interface.

…cessed.

…ructor. Refactored KafkaOffset constructors.

…ded missing hashCode() overrides and other minor fixes.

…ded missing default keywords to switches.

…factored Exceptions to conform to Checkstyle standards.

…ap conversion to HdfsOffset and KafkaOffset, solving code duplication of the conversion. Implemented additional OffsetInterface.java for HdfsOffset and KafkaOffset.

Tiihott · 2024-11-11T10:30:29Z

Rebased again to current main branch. Applied Checkstyle cleanup to the new files from rebase.

Tiihott self-assigned this Sep 20, 2024

Tiihott requested a review from eemhu September 23, 2024 08:18