Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

improved dependencies and connection handling #10

Merged
merged 2 commits into from
Dec 19, 2013
Merged

improved dependencies and connection handling #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
891d064
cleaned up dependencies
mleich Dec 19, 2013
3a5ca07
improced connect/disconnect/reconnect handlind
mleich Dec 19, 2013
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 42 additions & 12 deletions meteor-scheduler-client/src/main/java/eu/stratosphere/meteor/client/DOPAClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,36 +89,66 @@ public String getClientID(){
* Try to connect the client with the scheduler services.
* If this failed for any reason you can try it again.
*
* It do nothing if the client is still connected.
* It does nothing if the client is still connected.
*
* @return <code>true</code> if the client was successfully connected or the client was already connected,
* <code>false</code> if the client was not connected to the server
*/
public void connect() {
public boolean connect() {
// if the client is still connect
if ( this.connectionFac != null ) {
LOG.error( "The client is still connected. If you want to reconnect the client disconnect it first." );
return;
return true;
}

// else try to connect it
try { this.connectionFac = new ClientConnectionFactory( this ); }
catch ( Exception exc ) { LOG.error( "Cannot connected to the scheduler services!", exc ); }
try {
this.connectionFac = new ClientConnectionFactory( this );
return true;
}
catch ( Exception exc ) {
LOG.error( "Cannot connected to the scheduler services!", exc );
}
return false;
}

/**
* Try to reconnect the client with the scheduler services.
*
* Nothing happened if this client is still connected.
* Tries to reconnect the client with the scheduler services.
* Does nothing if this client is still connected.
* Use this method if you want to reconnect to the scheduler but didn't properly disconnect before.
* This is especially useful if you client crashed and the scheduler assumes it is still connected.
*
* ¡Use this method with great care!
* We only support one active client instance per client ID. Reconnect only if you are sure the previous
* client instance that created the connection is not running anymore!
*
* * @return <code>true</code> if the client was successfully reconnected or the client was already connected,
* <code>false</code> if the client was not reconnected to the server
*/
public void reconnect(){
public boolean reconnect(){
// if the client is still connect
if ( this.connectionFac != null ) {
LOG.error( "The client is still connected. If you want to reconnect the client disconnect it first." );
return;
return true;
}

// try to reconnect
try { this.connectionFac = new ClientConnectionFactory( this, true ); }
catch ( Exception exc ) { LOG.error("Cannot reconnect to the scheduler services!", exc); }
try { this.connectionFac = new ClientConnectionFactory( this, true );
return true;
}
catch ( Exception exc ) {
LOG.error("Cannot reconnect to the scheduler services!", exc);
}
return false;
}

/**
* Check whether this client is connected to the server
* @return <code>true</code> if this client is connected to the server
*/
public boolean isConnected () {
return this.connectionFac != null;
}

/**
* Try to disconnect the client.
Expand Down
34 changes: 28 additions & 6 deletions meteor-scheduler-client/src/test/java/eu/stratosphere/meteor/client/DOPAClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package eu.stratosphere.meteor.client;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import eu.stratosphere.meteor.common.DSCLJob;
import eu.stratosphere.meteor.common.JobState;
Expand All @@ -12,15 +14,33 @@ public class DOPAClientTest {
@Test
public void testConnectDisconnect () {
DOPAClient client = DOPAClient.createNewClient("connectDisconnectID");
client.connect();
assertTrue(client.connect());
client.disconnect();
}

@Test
public void testReconnect () {
// connect first client
DOPAClient client = DOPAClient.createNewClient("testReconnectID");
assertTrue("First client could not connect to scheduler", client.connect());
// leave connection open
client = null;

// connect second client with same ID
DOPAClient client2 = DOPAClient.createNewClient("testReconnectID");
// connection should fail as the first client didn't disconnect properly
assertFalse("Second client did connect to server, connection was expected to fail!", client2.connect());
// explicit reconnect should work
assertTrue("Reconnecting second client failed", client2.reconnect());
// do proper disconnect
client2.disconnect();
}


@Test
public void testQuerySubmission () {
DOPAClient client = DOPAClient.createNewClient("testID");
client.connect();
assertTrue("Could not connect client to scheduler", client.connect());
JobStateListener listener = new JobStateListener() {
@Override
public void stateChanged(DSCLJob job, JobState newStatus) {
Expand All @@ -31,18 +51,20 @@ public void stateChanged(DSCLJob job, JobState newStatus) {
DSCLJob job = client.createNewJob("PROVOKE SYNTAX ERROR", listener);

long start = System.currentTimeMillis();
boolean expectedErrorFlag = false;

// wait for 10.000ms = 10 seconds
while ( System.currentTimeMillis() - start < 10_000 ) {
if ( job.getStatus().equals(JobState.ERROR) )return;

if ( job.getStatus().equals(JobState.ERROR) ) {
expectedErrorFlag = true;
}
try {Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

fail("Job doesn't finished with an error in 10 seconds.");
client.disconnect();
assertTrue("Job doesn't finished with an error in 10 seconds.", expectedErrorFlag);
}

}
7 changes: 7 additions & 0 deletions meteor-scheduler-common/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,5 +11,12 @@

<artifactId>meteor-scheduler-common</artifactId>

<dependencies>
<dependency>
<groupId>org.json</groupId>
<artifactId>json</artifactId>
<version>20090211</version>
</dependency>
</dependencies>

</project>
16 changes: 16 additions & 0 deletions meteor-scheduler-server/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,22 @@
<artifactId>meteor-scheduler-common</artifactId>
<version>0.4-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>eu.stratosphere</groupId>
<artifactId>meteor-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>eu.stratosphere</groupId>
<artifactId>meteor-meteor</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>eu.stratosphere</groupId>
<artifactId>meteor-webfrontend</artifactId>
<version>${project.version}</version>
</dependency>

</dependencies>

</project>
39 changes: 10 additions & 29 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,55 +25,36 @@
<version>0.4-SNAPSHOT</version>

<dependencies>
<dependency>
<groupId>eu.stratosphere</groupId>
<artifactId>meteor-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>eu.stratosphere</groupId>
<artifactId>meteor-meteor</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>eu.stratosphere</groupId>
<artifactId>meteor-webfrontend</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.1.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-core</artifactId>
<version>0.20.205.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-core</artifactId>
<version>0.20.205.0</version>
</dependency>
<!-- <dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-junit4</artifactId>
<version>1.5</version>
<type>jar</type>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.json</groupId>
<artifactId>json</artifactId>
<version>20090211</version>
</dependency>
<dependency>
</dependency> -->
<!-- <dependency>
<groupId>com.github.stefanbirkner</groupId>
<artifactId>system-rules</artifactId>
<version>1.3.1</version>
<type>jar</type>
<scope>compile</scope>
</dependency>
</dependency> -->
</dependencies>

<reporting>
Expand Down