提交 50301254 编写于 作者: N Nico Kruber 提交者: zentol

[FLINK-8280][checkstyle] enable and fix checkstyle in BlobServer and BlobUtils

This closes #5175.
上级 3cdc5d1d
......@@ -119,7 +119,7 @@ public final class BlobClient implements Closeable {
}
}
catch(Exception e) {
catch (Exception e) {
BlobUtils.closeSilently(socket, LOG);
throw new IOException("Could not connect to BlobServer at address " + serverAddress, e);
}
......
......@@ -50,7 +50,7 @@ final class BlobInputStream extends InputStream {
private final OutputStream wrappedOutputStream;
/**
* The BLOB key if the GET operation has been performed on a content-addressable BLOB, otherwise <code>null<code>.
* The BLOB key if the GET operation has been performed on a content-addressable BLOB, otherwise <code>null</code>.
*/
private final BlobKey blobKey;
......
......@@ -77,28 +77,28 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
/** The server socket listening for incoming connections. */
private final ServerSocket serverSocket;
/** The SSL server context if ssl is enabled for the connections */
/** The SSL server context if ssl is enabled for the connections. */
private final SSLContext serverSSLContext;
/** Blob Server configuration */
/** Blob Server configuration. */
private final Configuration blobServiceConfiguration;
/** Indicates whether a shutdown of server component has been requested. */
private final AtomicBoolean shutdownRequested = new AtomicBoolean();
/** Root directory for local file storage */
/** Root directory for local file storage. */
private final File storageDir;
/** Blob store for distributed file storage, e.g. in HA */
/** Blob store for distributed file storage, e.g. in HA. */
private final BlobStore blobStore;
/** Set of currently running threads */
/** Set of currently running threads. */
private final Set<BlobServerConnection> activeConnections = new HashSet<>();
/** The maximum number of concurrent connections */
/** The maximum number of concurrent connections. */
private final int maxConnections;
/** Lock guarding concurrent file accesses */
/** Lock guarding concurrent file accesses. */
private final ReadWriteLock readWriteLock;
/**
......@@ -201,8 +201,8 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
}
});
if(socketAttempt == null) {
throw new IOException("Unable to allocate socket for blob server in specified port range: "+serverPortRange);
if (socketAttempt == null) {
throw new IOException("Unable to allocate socket for blob server in specified port range: " + serverPortRange);
} else {
SSLUtils.setSSLVerAndCipherSuites(socketAttempt, config);
this.serverSocket = socketAttempt;
......@@ -254,7 +254,7 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
}
/**
* Returns the lock used to guard file accesses
* Returns the lock used to guard file accesses.
*/
ReadWriteLock getReadWriteLock() {
return readWriteLock;
......@@ -360,7 +360,7 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
}
}
if(LOG.isInfoEnabled()) {
if (LOG.isInfoEnabled()) {
LOG.info("Stopped BLOB server at {}:{}", serverSocket.getInetAddress().getHostAddress(), getPort());
}
......@@ -375,8 +375,8 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
/**
* Retrieves the local path of a (job-unrelated) file associated with a job and a blob key.
* <p>
* The blob server looks the blob key up in its local storage. If the file exists, it is
*
* <p>The blob server looks the blob key up in its local storage. If the file exists, it is
* returned. If the file does not exist, it is retrieved from the HA blob store (if available)
* or a {@link FileNotFoundException} is thrown.
*
......@@ -395,8 +395,8 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
/**
* Retrieves the local path of a file associated with a job and a blob key.
* <p>
* The blob server looks the blob key up in its local storage. If the file exists, it is
*
* <p>The blob server looks the blob key up in its local storage. If the file exists, it is
* returned. If the file does not exist, it is retrieved from the HA blob store (if available)
* or a {@link FileNotFoundException} is thrown.
*
......@@ -419,8 +419,8 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
/**
* Returns the path to a local copy of the file associated with the provided job ID and blob
* key.
* <p>
* We will first attempt to serve the BLOB from the local storage. If the BLOB is not in
*
* <p>We will first attempt to serve the BLOB from the local storage. If the BLOB is not in
* there, we will try to download it from the HA store.
*
* @param jobId
......@@ -443,8 +443,8 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
/**
* Retrieves the local path of a file associated with a job and a blob key.
* <p>
* The blob server looks the blob key up in its local storage. If the file exists, it is
*
* <p>The blob server looks the blob key up in its local storage. If the file exists, it is
* returned. If the file does not exist, it is retrieved from the HA blob store (if available)
* or a {@link FileNotFoundException} is thrown.
*
......@@ -474,12 +474,12 @@ public class BlobServer extends Thread implements BlobService, BlobWriter, Perma
/**
* Helper to retrieve the local path of a file associated with a job and a blob key.
* <p>
* The blob server looks the blob key up in its local storage. If the file exists, it is
*
* <p>The blob server looks the blob key up in its local storage. If the file exists, it is
* returned. If the file does not exist, it is retrieved from the HA blob store (if available)
* or a {@link FileNotFoundException} is thrown.
* <p>
* <strong>Assumes the read lock has already been acquired.</strong>
*
* <p><strong>Assumes the read lock has already been acquired.</strong>
*
* @param jobId
* ID of the job this blob belongs to (or <tt>null</tt> if job-unrelated)
......
......@@ -15,8 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.runtime.blob;
/**
* Defines constants for the protocol between the BLOB {@link BlobServer server} and the
* {@link AbstractBlobCache caches}.
*/
public class BlobServerProtocol {
// --------------------------------------------------------------------------------------------
......
......@@ -150,8 +150,8 @@ public class BlobUtils {
File storageDir;
// NOTE: although we will be using UUIDs, there may be collisions
final int MAX_ATTEMPTS = 10;
for(int attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
int maxAttempts = 10;
for (int attempt = 0; attempt < maxAttempts; attempt++) {
storageDir = new File(baseDir, String.format(
"blobStore-%s", UUID.randomUUID().toString()));
......@@ -251,8 +251,8 @@ public class BlobUtils {
/**
* Returns the path for the given blob key.
* <p>
* The returned path can be used with the (local or HA) BLOB store file system back-end for
*
* <p>The returned path can be used with the (local or HA) BLOB store file system back-end for
* recovery purposes and follows the same scheme as {@link #getStorageLocation(File, JobID,
* BlobKey)}.
*
......@@ -403,12 +403,12 @@ public class BlobUtils {
}
}
static void closeSilently(Socket socket, Logger LOG) {
static void closeSilently(Socket socket, Logger log) {
if (socket != null) {
try {
socket.close();
} catch (Throwable t) {
LOG.debug("Exception while closing BLOB server connection socket.", t);
log.debug("Exception while closing BLOB server connection socket.", t);
}
}
}
......
......@@ -46,10 +46,10 @@ public class FileSystemBlobStore implements BlobStoreService {
private static final Logger LOG = LoggerFactory.getLogger(FileSystemBlobStore.class);
/** The file system in which blobs are stored */
/** The file system in which blobs are stored. */
private final FileSystem fileSystem;
/** The base path of the blob store */
/** The base path of the blob store. */
private final String basePath;
public FileSystemBlobStore(FileSystem fileSystem, String storagePath) throws IOException {
......
......@@ -78,7 +78,7 @@ import static org.mockito.Mockito.mock;
/**
* Tests for GET-specific parts of the {@link BlobCacheService}.
*
* This includes access of transient BLOBs from the {@link PermanentBlobCache}, permanent BLOBS from
* <p>This includes access of transient BLOBs from the {@link PermanentBlobCache}, permanent BLOBS from
* the {@link TransientBlobCache}, and how failing GET requests behave in the presence of failures
* when used with a {@link BlobCacheService}.
*
......
......@@ -36,10 +36,10 @@ import java.io.IOException;
public class BlobClientSslTest extends BlobClientTest {
/** The instance of the SSL BLOB server used during the tests. */
private static BlobServer BLOB_SSL_SERVER;
private static BlobServer blobSslServer;
/** Instance of a non-SSL BLOB server with SSL-enabled security options. */
private static BlobServer BLOB_NON_SSL_SERVER;
private static BlobServer blobNonSslServer;
/** The SSL blob service client configuration. */
private static Configuration sslClientConfig;
......@@ -62,8 +62,8 @@ public class BlobClientSslTest extends BlobClientTest {
config.setString(SecurityOptions.SSL_KEYSTORE, "src/test/resources/local127.keystore");
config.setString(SecurityOptions.SSL_KEYSTORE_PASSWORD, "password");
config.setString(SecurityOptions.SSL_KEY_PASSWORD, "password");
BLOB_SSL_SERVER = new BlobServer(config, new VoidBlobStore());
BLOB_SSL_SERVER.start();
blobSslServer = new BlobServer(config, new VoidBlobStore());
blobSslServer.start();
sslClientConfig = new Configuration();
sslClientConfig.setBoolean(SecurityOptions.SSL_ENABLED, true);
......@@ -81,8 +81,8 @@ public class BlobClientSslTest extends BlobClientTest {
config.setString(SecurityOptions.SSL_KEYSTORE, "src/test/resources/local127.keystore");
config.setString(SecurityOptions.SSL_KEYSTORE_PASSWORD, "password");
config.setString(SecurityOptions.SSL_KEY_PASSWORD, "password");
BLOB_NON_SSL_SERVER = new BlobServer(config, new VoidBlobStore());
BLOB_NON_SSL_SERVER.start();
blobNonSslServer = new BlobServer(config, new VoidBlobStore());
blobNonSslServer.start();
nonSslClientConfig = new Configuration();
nonSslClientConfig.setBoolean(SecurityOptions.SSL_ENABLED, true);
......@@ -96,11 +96,11 @@ public class BlobClientSslTest extends BlobClientTest {
*/
@AfterClass
public static void stopServers() throws IOException {
if (BLOB_SSL_SERVER != null) {
BLOB_SSL_SERVER.close();
if (blobSslServer != null) {
blobSslServer.close();
}
if (BLOB_NON_SSL_SERVER != null) {
BLOB_NON_SSL_SERVER.close();
if (blobNonSslServer != null) {
blobNonSslServer.close();
}
}
......@@ -109,7 +109,7 @@ public class BlobClientSslTest extends BlobClientTest {
}
protected BlobServer getBlobServer() {
return BLOB_SSL_SERVER;
return blobSslServer;
}
/**
......@@ -117,7 +117,7 @@ public class BlobClientSslTest extends BlobClientTest {
*/
@Test
public void testUploadJarFilesHelper() throws Exception {
uploadJarFile(BLOB_SSL_SERVER, sslClientConfig);
uploadJarFile(blobSslServer, sslClientConfig);
}
/**
......@@ -126,7 +126,7 @@ public class BlobClientSslTest extends BlobClientTest {
@Test(expected = IOException.class)
public void testSSLClientFailure() throws Exception {
// SSL client connected to non-ssl server
uploadJarFile(BLOB_SERVER, sslClientConfig);
uploadJarFile(blobServer, sslClientConfig);
}
/**
......@@ -135,7 +135,7 @@ public class BlobClientSslTest extends BlobClientTest {
@Test(expected = IOException.class)
public void testSSLClientFailure2() throws Exception {
// SSL client connected to non-ssl server
uploadJarFile(BLOB_NON_SSL_SERVER, sslClientConfig);
uploadJarFile(blobNonSslServer, sslClientConfig);
}
/**
......@@ -144,7 +144,7 @@ public class BlobClientSslTest extends BlobClientTest {
@Test(expected = IOException.class)
public void testSSLServerFailure() throws Exception {
// Non-SSL client connected to ssl server
uploadJarFile(BLOB_SSL_SERVER, clientConfig);
uploadJarFile(blobSslServer, clientConfig);
}
/**
......@@ -153,7 +153,7 @@ public class BlobClientSslTest extends BlobClientTest {
@Test(expected = IOException.class)
public void testSSLServerFailure2() throws Exception {
// Non-SSL client connected to ssl server
uploadJarFile(BLOB_SSL_SERVER, nonSslClientConfig);
uploadJarFile(blobSslServer, nonSslClientConfig);
}
/**
......@@ -161,7 +161,7 @@ public class BlobClientSslTest extends BlobClientTest {
*/
@Test
public void testNonSSLConnection() throws Exception {
uploadJarFile(BLOB_SERVER, clientConfig);
uploadJarFile(blobServer, clientConfig);
}
/**
......@@ -169,7 +169,7 @@ public class BlobClientSslTest extends BlobClientTest {
*/
@Test
public void testNonSSLConnection2() throws Exception {
uploadJarFile(BLOB_SERVER, nonSslClientConfig);
uploadJarFile(blobServer, nonSslClientConfig);
}
/**
......@@ -177,7 +177,7 @@ public class BlobClientSslTest extends BlobClientTest {
*/
@Test
public void testNonSSLConnection3() throws Exception {
uploadJarFile(BLOB_NON_SSL_SERVER, clientConfig);
uploadJarFile(blobNonSslServer, clientConfig);
}
/**
......@@ -185,6 +185,6 @@ public class BlobClientSslTest extends BlobClientTest {
*/
@Test
public void testNonSSLConnection4() throws Exception {
uploadJarFile(BLOB_NON_SSL_SERVER, nonSslClientConfig);
uploadJarFile(blobNonSslServer, nonSslClientConfig);
}
}
......@@ -62,7 +62,7 @@ public class BlobClientTest extends TestLogger {
private static final int TEST_BUFFER_SIZE = 17 * 1000;
/** The instance of the (non-ssl) BLOB server used during the tests. */
static BlobServer BLOB_SERVER;
static BlobServer blobServer;
/** The blob service (non-ssl) client configuration. */
static Configuration clientConfig;
......@@ -79,8 +79,8 @@ public class BlobClientTest extends TestLogger {
config.setString(BlobServerOptions.STORAGE_DIRECTORY,
temporaryFolder.newFolder().getAbsolutePath());
BLOB_SERVER = new BlobServer(config, new VoidBlobStore());
BLOB_SERVER.start();
blobServer = new BlobServer(config, new VoidBlobStore());
blobServer.start();
clientConfig = new Configuration();
}
......@@ -90,8 +90,8 @@ public class BlobClientTest extends TestLogger {
*/
@AfterClass
public static void stopServer() throws IOException {
if (BLOB_SERVER != null) {
BLOB_SERVER.close();
if (blobServer != null) {
blobServer.close();
}
}
......@@ -319,7 +319,7 @@ public class BlobClientTest extends TestLogger {
}
protected BlobServer getBlobServer() {
return BLOB_SERVER;
return blobServer;
}
@Test
......
......@@ -78,7 +78,7 @@ public class TestingFailingBlobServer extends BlobServer {
if (socket != null) {
try {
socket.close();
} catch(Throwable ignored) {}
} catch (Throwable ignored) {}
}
}
}
......
......@@ -23,13 +23,6 @@ under the License.
"http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress
files="(.*)runtime[/\\]blob[/\\](.*)"
checks="NewlineAtEndOfFile|RegexpSingleline|TodoComment|RedundantImport|ImportOrder|RedundantModifier|JavadocMethod|JavadocParagraph|JavadocType|JavadocStyle|PackageName|TypeNameCheck|ConstantNameCheck|StaticVariableNameCheck|MemberNameCheck|MethodNameCheck|ParameterName|LocalFinalVariableName|LocalVariableName|LeftCurly|UpperEll|FallThrough|reliefPattern|SimplifyBooleanExpression|EmptyStatement|ModifierOrder|EmptyLineSeparator|WhitespaceAround|WhitespaceAfter|NoWhitespaceAfter|NoWhitespaceBefore|OperatorWrap|ParenPad"/>
<!--Only additional checks for test sources. Those checks were present in the "pre-strict" checkstyle but were not applied to test sources. We do not want to suppress them for sources directory-->
<suppress
files="(.*)test[/\\](.*)runtime[/\\]blob[/\\](.*)"
checks="AvoidStarImport|UnusedImport"/>
<suppress
files="(.*)runtime[/\\]checkpoint[/\\](.*)"
checks="NewlineAtEndOfFile|RegexpSingleline|TodoComment|RedundantImport|ImportOrder|RedundantModifier|JavadocMethod|JavadocParagraph|JavadocType|JavadocStyle|PackageName|TypeNameCheck|ConstantNameCheck|StaticVariableNameCheck|MemberNameCheck|MethodNameCheck|ParameterName|LocalFinalVariableName|LocalVariableName|LeftCurly|UpperEll|FallThrough|reliefPattern|SimplifyBooleanExpression|EmptyStatement|ModifierOrder|EmptyLineSeparator|WhitespaceAround|WhitespaceAfter|NoWhitespaceAfter|NoWhitespaceBefore|OperatorWrap|ParenPad"/>
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册