Skip to content
Snippets Groups Projects
Commit 74aa0df8 authored by Tathagata Das's avatar Tathagata Das
Browse files

[SPARK-20377][SS] Fix JavaStructuredSessionization example

## What changes were proposed in this pull request?

Extra accessors in java bean class causes incorrect encoder generation, which corrupted the state when using timeouts.

## How was this patch tested?
manually ran the example

Author: Tathagata Das <tathagata.das1565@gmail.com>

Closes #17676 from tdas/SPARK-20377.
parent f654b39a
No related branches found
No related tags found
No related merge requests found
......@@ -76,8 +76,6 @@ public final class JavaStructuredSessionization {
for (String word : lineWithTimestamp.getLine().split(" ")) {
eventList.add(new Event(word, lineWithTimestamp.getTimestamp()));
}
System.out.println(
"Number of events from " + lineWithTimestamp.getLine() + " = " + eventList.size());
return eventList.iterator();
}
};
......@@ -100,7 +98,7 @@ public final class JavaStructuredSessionization {
// If timed out, then remove session and send final update
if (state.hasTimedOut()) {
SessionUpdate finalUpdate = new SessionUpdate(
sessionId, state.get().getDurationMs(), state.get().getNumEvents(), true);
sessionId, state.get().calculateDuration(), state.get().getNumEvents(), true);
state.remove();
return finalUpdate;
......@@ -133,7 +131,7 @@ public final class JavaStructuredSessionization {
// Set timeout such that the session will be expired if no data received for 10 seconds
state.setTimeoutDuration("10 seconds");
return new SessionUpdate(
sessionId, state.get().getDurationMs(), state.get().getNumEvents(), false);
sessionId, state.get().calculateDuration(), state.get().getNumEvents(), false);
}
}
};
......@@ -215,7 +213,8 @@ public final class JavaStructuredSessionization {
public long getEndTimestampMs() { return endTimestampMs; }
public void setEndTimestampMs(long endTimestampMs) { this.endTimestampMs = endTimestampMs; }
public long getDurationMs() { return endTimestampMs - startTimestampMs; }
public long calculateDuration() { return endTimestampMs - startTimestampMs; }
@Override public String toString() {
return "SessionInfo(numEvents = " + numEvents +
", timestamps = " + startTimestampMs + " to " + endTimestampMs + ")";
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment