<commit_before>/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.emc.pravega.service.storage.impl.hdfs;

import com.emc.pravega.service.contracts.SegmentProperties;
import com.emc.pravega.service.contracts.StreamSegmentInformation;
import com.emc.pravega.service.storage.BadOffsetException;
import com.emc.pravega.service.storage.Storage;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;

import java.io.IOException;
import java.io.InputStream;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

@Slf4j
class HDFSLowerStorage implements Storage {
    private final Executor executor;
    private FileSystem fs;
    private Configuration conf;
    private final HDFSStorageConfig serviceBuilderConfig;

    public HDFSLowerStorage(HDFSStorageConfig serviceBuilderConfig, Executor executor) {
        this.serviceBuilderConfig = serviceBuilderConfig;
        this.executor = executor;
    }

    @Override
    public CompletableFuture<SegmentProperties> create(String streamSegmentName, Duration timeout) {
        CompletableFuture<SegmentProperties> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
                    try {
                        retVal.complete(createSync(streamSegmentName, timeout));
                    } catch (IOException e) {
                        retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
                    }
                },
                executor);
        return retVal;
    }


    SegmentProperties createSync(String streamSegmentName, Duration timeout) throws IOException {
            getFS().create(new Path(streamSegmentName),
                    new FsPermission(FsAction.READ_WRITE, FsAction.NONE, FsAction.NONE),
                    false,
                    0,
                    this.serviceBuilderConfig.getReplication(),
                    this.serviceBuilderConfig.getBlockSize(),
                    null).close();
            return new StreamSegmentInformation(streamSegmentName,
                    0,
                    false,
                    false,
                    new Date()
            );
    }

    SegmentProperties getStreamSegmentInfoSync(String streamSegmentName, Duration timeout) throws IOException {
        FileStatus[] status = getFS().globStatus(new Path(streamSegmentName));
        return new StreamSegmentInformation(streamSegmentName,
                status[0].getLen(),
                status[0].getPermission().getUserAction() == FsAction.READ,
                false,
                new Date(status[0].getModificationTime()));

    }


    @Override
    public CompletableFuture<Boolean> acquireLockForSegment(String streamSegmentName) {
        //Acquirelock is not implemented at the lower level of HDFS
        CompletableFuture<Boolean> retVal = new CompletableFuture<>();
        retVal.completeExceptionally(new IOException("Not Implemented"));
        return retVal;
    }

    @Override
    public CompletableFuture<Void> write(String streamSegmentName, long offset, InputStream data, int length, Duration timeout) {
        CompletableFuture<Void> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
                    try {
                        retVal.complete(writeSync(streamSegmentName, offset, length, data, timeout));
                    } catch (Exception e) {
                        retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
                    }
                },
                executor);
        return retVal;
    }



    private Void writeSync(String streamSegmentName, long offset, int length, InputStream data, Duration timeout)
            throws BadOffsetException, IOException {
        try (FSDataOutputStream stream = getFS().append(new Path(streamSegmentName))) {
        if (stream.getPos() != offset) {
            throw new BadOffsetException("Offset in the stream segment " + streamSegmentName + "" +
                    " already has some data beyond offset "+ offset);
        }
            IOUtils.copyBytes(data, stream, length);
            stream.flush();
        }
        return null;
    }


    @Override
    public CompletableFuture<SegmentProperties> seal(String streamSegmentName, Duration timeout) {
        CompletableFuture<SegmentProperties> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
                    try {
                        retVal.complete(sealSync(streamSegmentName, timeout));
                    } catch (IOException e) {
                        retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
                    }
                },
                executor);
        return retVal;
    }

    SegmentProperties sealSync(String streamSegmentName, Duration timeout) throws IOException {
        getFS().setPermission(
                new Path(streamSegmentName),
                new FsPermission(FsAction.READ,
                        FsAction.NONE,
                        FsAction.NONE
                )
        );
        return this.getStreamSegmentInfoSync(streamSegmentName, timeout);
    }

    @Override
    public CompletableFuture<Void> concat(String targetStreamSegmentName, long offset, String sourceStreamSegmentName, Duration timeout) {
        CompletableFuture<Void> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
            try {
                retVal.complete(concatSync(targetStreamSegmentName, offset, sourceStreamSegmentName, timeout));
            } catch (Exception e) {
                retVal.completeExceptionally(ExceptionHelper.translateFromIOException(sourceStreamSegmentName, e));
            }
        }, executor);
        return retVal;
    }


    Void concatSync(String targetStreamSegmentName, long offset, String sourceStreamSegmentName, Duration timeout) throws IOException, BadOffsetException {
        if (getFS().globStatus(new Path(targetStreamSegmentName))[0].getLen() != offset )
            throw new BadOffsetException(targetStreamSegmentName + " has more data than" + offset);
        getFS().concat(new Path(targetStreamSegmentName),
                new Path[]{
                        new Path(targetStreamSegmentName),
                        new Path(sourceStreamSegmentName)
                });
        return null;
    }

    @Override
    public CompletableFuture<Void> delete(String streamSegmentName, Duration timeout) {
        CompletableFuture<Void> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
            try {
                retVal.complete(deleteSync(streamSegmentName, timeout));
            } catch (IOException e) {
                retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
            }
        }, executor);
        return retVal;
    }

    @Override
    public void close() {<commit_msg>Pull from master (checkstyle changes)<commit_after>/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.emc.pravega.service.storage.impl.hdfs;

import com.emc.pravega.service.contracts.SegmentProperties;
import com.emc.pravega.service.contracts.StreamSegmentInformation;
import com.emc.pravega.service.storage.BadOffsetException;
import com.emc.pravega.service.storage.Storage;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;

import java.io.IOException;
import java.io.InputStream;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

@Slf4j
class HDFSLowerStorage implements Storage {
    private final Executor executor;
    private FileSystem fs;
    private Configuration conf;
    private final HDFSStorageConfig serviceBuilderConfig;

    public HDFSLowerStorage(HDFSStorageConfig serviceBuilderConfig, Executor executor) {
        this.serviceBuilderConfig = serviceBuilderConfig;
        this.executor = executor;
    }

    @Override
    public CompletableFuture<SegmentProperties> create(String streamSegmentName, Duration timeout) {
        CompletableFuture<SegmentProperties> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
                    try {
                        retVal.complete(createSync(streamSegmentName, timeout));
                    } catch (IOException e) {
                        retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
                    }
                },
                executor);
        return retVal;
    }


    SegmentProperties createSync(String streamSegmentName, Duration timeout) throws IOException {
            getFS().create(new Path(streamSegmentName),
                    new FsPermission(FsAction.READ_WRITE, FsAction.NONE, FsAction.NONE),
                    false,
                    0,
                    this.serviceBuilderConfig.getReplication(),
                    this.serviceBuilderConfig.getBlockSize(),
                    null).close();
            return new StreamSegmentInformation(streamSegmentName,
                    0,
                    false,
                    false,
                    new Date()
            );
    }

    SegmentProperties getStreamSegmentInfoSync(String streamSegmentName, Duration timeout) throws IOException {
        FileStatus[] status = getFS().globStatus(new Path(streamSegmentName));
        return new StreamSegmentInformation(streamSegmentName,
                status[0].getLen(),
                status[0].getPermission().getUserAction() == FsAction.READ,
                false,
                new Date(status[0].getModificationTime()));

    }


    @Override
    public CompletableFuture<Boolean> acquireLockForSegment(String streamSegmentName) {
        //Acquirelock is not implemented at the lower level of HDFS
        CompletableFuture<Boolean> retVal = new CompletableFuture<>();
        retVal.completeExceptionally(new IOException("Not Implemented"));
        return retVal;
    }

    @Override
    public CompletableFuture<Void> write(String streamSegmentName, long offset, InputStream data, int length, Duration timeout) {
        CompletableFuture<Void> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
                    try {
                        retVal.complete(writeSync(streamSegmentName, offset, length, data, timeout));
                    } catch (Exception e) {
                        retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
                    }
                },
                executor);
        return retVal;
    }



    private Void writeSync(String streamSegmentName, long offset, int length, InputStream data, Duration timeout)
            throws BadOffsetException, IOException {
        try (FSDataOutputStream stream = getFS().append(new Path(streamSegmentName))) {
        if (stream.getPos() != offset) {
            throw new BadOffsetException("Offset in the stream segment " + streamSegmentName + "" +
                    " already has some data beyond offset "+ offset);
        }
            IOUtils.copyBytes(data, stream, length);
            stream.flush();
        }
        return null;
    }


    @Override
    public CompletableFuture<SegmentProperties> seal(String streamSegmentName, Duration timeout) {
        CompletableFuture<SegmentProperties> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
                    try {
                        retVal.complete(sealSync(streamSegmentName, timeout));
                    } catch (IOException e) {
                        retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
                    }
                },
                executor);
        return retVal;
    }

    SegmentProperties sealSync(String streamSegmentName, Duration timeout) throws IOException {
        getFS().setPermission(
                new Path(streamSegmentName),
                new FsPermission(FsAction.READ,
                        FsAction.NONE,
                        FsAction.NONE
                )
        );
        return this.getStreamSegmentInfoSync(streamSegmentName, timeout);
    }

    @Override
    public CompletableFuture<Void> concat(String targetStreamSegmentName, long offset, String sourceStreamSegmentName, Duration timeout) {
        CompletableFuture<Void> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
            try {
                retVal.complete(concatSync(targetStreamSegmentName, offset, sourceStreamSegmentName, timeout));
            } catch (Exception e) {
                retVal.completeExceptionally(ExceptionHelper.translateFromIOException(sourceStreamSegmentName, e));
            }
        }, executor);
        return retVal;
    }


    Void concatSync(String targetStreamSegmentName, long offset, String sourceStreamSegmentName, Duration timeout) throws IOException, BadOffsetException {
        if (getFS().globStatus(new Path(targetStreamSegmentName))[0].getLen() != offset ) {
            throw new BadOffsetException(targetStreamSegmentName + " has more data than" + offset);
        }
        getFS().concat(new Path(targetStreamSegmentName),
                new Path[]{
                        new Path(targetStreamSegmentName),
                        new Path(sourceStreamSegmentName)
                });
        return null;
    }

    @Override
    public CompletableFuture<Void> delete(String streamSegmentName, Duration timeout) {
        CompletableFuture<Void> retVal = new CompletableFuture<>();
        CompletableFuture.runAsync(() -> {
            try {
                retVal.complete(deleteSync(streamSegmentName, timeout));
            } catch (IOException e) {
                retVal.completeExceptionally(ExceptionHelper.translateFromIOException(streamSegmentName, e));
            }
        }, executor);
        return retVal;
    }

    @Override
    public void close() {
{'fraction_non_alphanumeric': 0.08289841000121374, 'fraction_numerical': 0.001577861390945503, 'mean_word_length': 2.520401623584704, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1571349', 'n_tokens_mistral': 4077, 'n_tokens_neox': 3916, 'n_words': 972}