You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mj...@apache.org on 2016/02/07 19:41:57 UTC
[1/2] flink git commit: [FLINK-2721] [Storm Compatibility] Add Tuple
meta information
Repository: flink
Updated Branches:
refs/heads/master 168c1f0a1 -> 7eb6edd57
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/StormTupleTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/StormTupleTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/StormTupleTest.java
index 595bada..eba611e 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/StormTupleTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/StormTupleTest.java
@@ -19,9 +19,10 @@ package org.apache.flink.storm.wrappers;
import backtype.storm.generated.GlobalStreamId;
import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
import backtype.storm.tuple.Values;
+
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.storm.util.AbstractTest;
import org.junit.Assert;
@@ -51,7 +52,8 @@ public class StormTupleTest extends AbstractTest {
public void nonTupleTest() {
final Object flinkTuple = this.r.nextInt();
- final StormTuple<Object> tuple = new StormTuple<Object>(flinkTuple, null);
+ final StormTuple<Object> tuple = new StormTuple<Object>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertSame(flinkTuple, tuple.getValue(0));
final List<Object> values = tuple.getValues();
@@ -61,24 +63,50 @@ public class StormTupleTest extends AbstractTest {
@Test
public void tupleTest() throws InstantiationException, IllegalAccessException {
- final int numberOfAttributes = this.r.nextInt(26);
- final Object[] data = new Object[numberOfAttributes];
+ for (int numberOfAttributes = 0; numberOfAttributes < 26; ++numberOfAttributes) {
+ final Object[] data = new Object[numberOfAttributes];
- final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance();
- for (int i = 0; i < numberOfAttributes; ++i) {
- data[i] = this.r.nextInt();
- flinkTuple.setField(data[i], i);
- }
+ final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance();
+ for (int i = 0; i < numberOfAttributes; ++i) {
+ data[i] = this.r.nextInt();
+ flinkTuple.setField(data[i], i);
+ }
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
- final List<Object> values = tuple.getValues();
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
+ final List<Object> values = tuple.getValues();
+
+ Assert.assertEquals(numberOfAttributes, values.size());
+ for (int i = 0; i < numberOfAttributes; ++i) {
+ Assert.assertEquals(flinkTuple.getField(i), values.get(i));
+ }
- Assert.assertEquals(numberOfAttributes, values.size());
- for (int i = 0; i < numberOfAttributes; ++i) {
- Assert.assertEquals(flinkTuple.getField(i), values.get(i));
+ Assert.assertEquals(numberOfAttributes, tuple.size());
}
+ }
+
+ @Test
+ public void tupleTestWithTaskId() throws InstantiationException, IllegalAccessException {
+ for (int numberOfAttributes = 1; numberOfAttributes < 26; ++numberOfAttributes) {
+ final Object[] data = new Object[numberOfAttributes];
+
+ final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance();
+ for (int i = 0; i < numberOfAttributes - 1; ++i) {
+ data[i] = this.r.nextInt();
+ flinkTuple.setField(data[i], i);
+ }
+
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, 0, null, null,
+ null);
+ final List<Object> values = tuple.getValues();
+
+ Assert.assertEquals(numberOfAttributes - 1, values.size());
+ for (int i = 0; i < numberOfAttributes - 1; ++i) {
+ Assert.assertEquals(flinkTuple.getField(i), values.get(i));
+ }
- Assert.assertEquals(numberOfAttributes, tuple.size());
+ Assert.assertEquals(numberOfAttributes - 1, tuple.size());
+ }
}
@Test
@@ -90,7 +118,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getBinary(index));
}
@@ -98,7 +127,8 @@ public class StormTupleTest extends AbstractTest {
public void testBoolean() {
final Boolean flinkTuple = this.r.nextBoolean();
- final StormTuple<Boolean> tuple = new StormTuple<Boolean>(flinkTuple, null);
+ final StormTuple<Boolean> tuple = new StormTuple<Boolean>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple, tuple.getBoolean(0));
}
@@ -106,7 +136,7 @@ public class StormTupleTest extends AbstractTest {
public void testByte() {
final Byte flinkTuple = (byte) this.r.nextInt();
- final StormTuple<Byte> tuple = new StormTuple<Byte>(flinkTuple, null);
+ final StormTuple<Byte> tuple = new StormTuple<Byte>(flinkTuple, null, -1, null, null, null);
Assert.assertEquals(flinkTuple, tuple.getByte(0));
}
@@ -114,7 +144,8 @@ public class StormTupleTest extends AbstractTest {
public void testDouble() {
final Double flinkTuple = this.r.nextDouble();
- final StormTuple<Double> tuple = new StormTuple<Double>(flinkTuple, null);
+ final StormTuple<Double> tuple = new StormTuple<Double>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple, tuple.getDouble(0));
}
@@ -122,7 +153,8 @@ public class StormTupleTest extends AbstractTest {
public void testFloat() {
final Float flinkTuple = this.r.nextFloat();
- final StormTuple<Float> tuple = new StormTuple<Float>(flinkTuple, null);
+ final StormTuple<Float> tuple = new StormTuple<Float>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple, tuple.getFloat(0));
}
@@ -130,7 +162,8 @@ public class StormTupleTest extends AbstractTest {
public void testInteger() {
final Integer flinkTuple = this.r.nextInt();
- final StormTuple<Integer> tuple = new StormTuple<Integer>(flinkTuple, null);
+ final StormTuple<Integer> tuple = new StormTuple<Integer>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple, tuple.getInteger(0));
}
@@ -138,7 +171,7 @@ public class StormTupleTest extends AbstractTest {
public void testLong() {
final Long flinkTuple = this.r.nextLong();
- final StormTuple<Long> tuple = new StormTuple<Long>(flinkTuple, null);
+ final StormTuple<Long> tuple = new StormTuple<Long>(flinkTuple, null, -1, null, null, null);
Assert.assertEquals(flinkTuple, tuple.getLong(0));
}
@@ -146,7 +179,8 @@ public class StormTupleTest extends AbstractTest {
public void testShort() {
final Short flinkTuple = (short) this.r.nextInt();
- final StormTuple<Short> tuple = new StormTuple<Short>(flinkTuple, null);
+ final StormTuple<Short> tuple = new StormTuple<Short>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple, tuple.getShort(0));
}
@@ -156,7 +190,8 @@ public class StormTupleTest extends AbstractTest {
this.r.nextBytes(data);
final String flinkTuple = new String(data);
- final StormTuple<String> tuple = new StormTuple<String>(flinkTuple, null);
+ final StormTuple<String> tuple = new StormTuple<String>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple, tuple.getString(0));
}
@@ -169,7 +204,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getBinary(index));
}
@@ -181,7 +217,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getBoolean(index));
}
@@ -193,7 +230,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getByte(index));
}
@@ -205,7 +243,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getDouble(index));
}
@@ -217,7 +256,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getFloat(index));
}
@@ -229,7 +269,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getInteger(index));
}
@@ -241,7 +282,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getLong(index));
}
@@ -253,7 +295,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getShort(index));
}
@@ -267,7 +310,8 @@ public class StormTupleTest extends AbstractTest {
final Tuple flinkTuple = new Tuple5<Object, Object, Object, Object, Object>();
flinkTuple.setField(data, index);
- final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null);
+ final StormTuple<Tuple> tuple = new StormTuple<Tuple>(flinkTuple, null, -1, null, null,
+ null);
Assert.assertEquals(flinkTuple.getField(index), tuple.getString(index));
}
@@ -275,7 +319,7 @@ public class StormTupleTest extends AbstractTest {
public void testContains() throws Exception {
Fields schema = new Fields("a1", "a2");
StormTuple<Object> tuple = new StormTuple<Object>(Tuple.getTupleClass(1).newInstance(),
- schema);
+ schema, -1, null, null, null);
Assert.assertTrue(tuple.contains("a1"));
Assert.assertTrue(tuple.contains("a2"));
@@ -285,11 +329,8 @@ public class StormTupleTest extends AbstractTest {
@Test
public void testGetFields() throws Exception {
Fields schema = new Fields();
-
- Assert.assertSame(schema, new StormTuple<Object>(Tuple.getTupleClass(1).newInstance(),
- schema).getFields());
-
- Assert.assertSame(null, new StormTuple<Object>(null, schema).getFields());
+ Assert.assertSame(schema,
+ new StormTuple<Object>(null, schema, -1, null, null, null).getFields());
}
@@ -297,13 +338,12 @@ public class StormTupleTest extends AbstractTest {
public void testFieldIndex() throws Exception {
Fields schema = new Fields("a1", "a2");
StormTuple<Object> tuple = new StormTuple<Object>(Tuple.getTupleClass(1).newInstance(),
- schema);
+ schema, -1, null, null, null);
Assert.assertEquals(0, tuple.fieldIndex("a1"));
Assert.assertEquals(1, tuple.fieldIndex("a2"));
}
- @SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testSelect() throws Exception {
Tuple tuple = Tuple.getTupleClass(arity).newInstance();
@@ -324,14 +364,14 @@ public class StormTupleTest extends AbstractTest {
Fields schema = new Fields(attributeNames);
Fields selector = new Fields(filterNames);
- Assert.assertEquals(values, new StormTuple(tuple, schema).select(selector));
+ Assert.assertEquals(values,
+ new StormTuple<>(tuple, schema, -1, null, null, null).select(selector));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetValueByField() throws Exception {
Object value = mock(Object.class);
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getValueByField(fieldName));
}
@@ -341,7 +381,7 @@ public class StormTupleTest extends AbstractTest {
Object value = mock(Object.class);
TestPojoMember<Object> pojo = new TestPojoMember<Object>(value);
StormTuple<TestPojoMember<Object>> tuple = new StormTuple<TestPojoMember<Object>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getValueByField(fieldNamePojo));
}
@@ -350,15 +390,14 @@ public class StormTupleTest extends AbstractTest {
Object value = mock(Object.class);
TestPojoGetter<Object> pojo = new TestPojoGetter<Object>(value);
StormTuple<TestPojoGetter<Object>> tuple = new StormTuple<TestPojoGetter<Object>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getValueByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetStringByField() throws Exception {
String value = "stringValue";
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getStringByField(fieldName));
}
@@ -367,7 +406,7 @@ public class StormTupleTest extends AbstractTest {
String value = "stringValue";
TestPojoMember<String> pojo = new TestPojoMember<String>(value);
StormTuple<TestPojoMember<String>> tuple = new StormTuple<TestPojoMember<String>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getStringByField(fieldNamePojo));
}
@@ -376,15 +415,14 @@ public class StormTupleTest extends AbstractTest {
String value = "stringValue";
TestPojoGetter<String> pojo = new TestPojoGetter<String>(value);
StormTuple<TestPojoGetter<String>> tuple = new StormTuple<TestPojoGetter<String>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getStringByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetIntegerByField() throws Exception {
Integer value = r.nextInt();
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getIntegerByField(fieldName));
}
@@ -393,7 +431,7 @@ public class StormTupleTest extends AbstractTest {
Integer value = r.nextInt();
TestPojoMember<Integer> pojo = new TestPojoMember<Integer>(value);
StormTuple<TestPojoMember<Integer>> tuple = new StormTuple<TestPojoMember<Integer>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getIntegerByField(fieldNamePojo));
}
@@ -402,15 +440,14 @@ public class StormTupleTest extends AbstractTest {
Integer value = r.nextInt();
TestPojoGetter<Integer> pojo = new TestPojoGetter<Integer>(value);
StormTuple<TestPojoGetter<Integer>> tuple = new StormTuple<TestPojoGetter<Integer>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getIntegerByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetLongByField() throws Exception {
Long value = r.nextLong();
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getLongByField(fieldName));
}
@@ -418,8 +455,8 @@ public class StormTupleTest extends AbstractTest {
public void testGetLongByFieldPojo() throws Exception {
Long value = r.nextLong();
TestPojoMember<Long> pojo = new TestPojoMember<Long>(value);
- StormTuple<TestPojoMember<Long>> tuple = new StormTuple<TestPojoMember<Long>>(pojo,
- null);
+ StormTuple<TestPojoMember<Long>> tuple = new StormTuple<TestPojoMember<Long>>(pojo, null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getLongByField(fieldNamePojo));
}
@@ -427,16 +464,15 @@ public class StormTupleTest extends AbstractTest {
public void testGetLongByFieldPojoGetter() throws Exception {
Long value = r.nextLong();
TestPojoGetter<Long> pojo = new TestPojoGetter<Long>(value);
- StormTuple<TestPojoGetter<Long>> tuple = new StormTuple<TestPojoGetter<Long>>(pojo,
- null);
+ StormTuple<TestPojoGetter<Long>> tuple = new StormTuple<TestPojoGetter<Long>>(pojo, null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getLongByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetBooleanByField() throws Exception {
Boolean value = r.nextBoolean();
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertEquals(value, tuple.getBooleanByField(fieldName));
}
@@ -445,7 +481,7 @@ public class StormTupleTest extends AbstractTest {
Boolean value = r.nextBoolean();
TestPojoMember<Boolean> pojo = new TestPojoMember<Boolean>(value);
StormTuple<TestPojoMember<Boolean>> tuple = new StormTuple<TestPojoMember<Boolean>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getBooleanByField(fieldNamePojo));
}
@@ -454,15 +490,14 @@ public class StormTupleTest extends AbstractTest {
Boolean value = r.nextBoolean();
TestPojoGetter<Boolean> pojo = new TestPojoGetter<Boolean>(value);
StormTuple<TestPojoGetter<Boolean>> tuple = new StormTuple<TestPojoGetter<Boolean>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getBooleanByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetShortByField() throws Exception {
Short value = (short) r.nextInt();
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getShortByField(fieldName));
}
@@ -471,7 +506,8 @@ public class StormTupleTest extends AbstractTest {
Short value = (short) r.nextInt();
TestPojoMember<Short> pojo = new TestPojoMember<Short>(value);
StormTuple<TestPojoMember<Short>> tuple = new StormTuple<TestPojoMember<Short>>(pojo,
- null);
+ null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getShortByField(fieldNamePojo));
}
@@ -480,15 +516,15 @@ public class StormTupleTest extends AbstractTest {
Short value = (short) r.nextInt();
TestPojoGetter<Short> pojo = new TestPojoGetter<Short>(value);
StormTuple<TestPojoGetter<Short>> tuple = new StormTuple<TestPojoGetter<Short>>(pojo,
- null);
+ null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getShortByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetByteByField() throws Exception {
Byte value = new Byte((byte) r.nextInt());
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getByteByField(fieldName));
}
@@ -497,7 +533,8 @@ public class StormTupleTest extends AbstractTest {
Byte value = new Byte((byte) r.nextInt());
TestPojoMember<Byte> pojo = new TestPojoMember<Byte>(value);
StormTuple<TestPojoMember<Byte>> tuple = new StormTuple<TestPojoMember<Byte>>(pojo,
- null);
+ null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getByteByField(fieldNamePojo));
}
@@ -506,15 +543,15 @@ public class StormTupleTest extends AbstractTest {
Byte value = new Byte((byte) r.nextInt());
TestPojoGetter<Byte> pojo = new TestPojoGetter<Byte>(value);
StormTuple<TestPojoGetter<Byte>> tuple = new StormTuple<TestPojoGetter<Byte>>(pojo,
- null);
+ null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getByteByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetDoubleByField() throws Exception {
Double value = r.nextDouble();
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getDoubleByField(fieldName));
}
@@ -523,7 +560,7 @@ public class StormTupleTest extends AbstractTest {
Double value = r.nextDouble();
TestPojoMember<Double> pojo = new TestPojoMember<Double>(value);
StormTuple<TestPojoMember<Double>> tuple = new StormTuple<TestPojoMember<Double>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getDoubleByField(fieldNamePojo));
}
@@ -532,15 +569,14 @@ public class StormTupleTest extends AbstractTest {
Double value = r.nextDouble();
TestPojoGetter<Double> pojo = new TestPojoGetter<Double>(value);
StormTuple<TestPojoGetter<Double>> tuple = new StormTuple<TestPojoGetter<Double>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(value, tuple.getDoubleByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetFloatByField() throws Exception {
Float value = r.nextFloat();
- StormTuple tuple = testGetByField(arity, index, value);
+ StormTuple<?> tuple = testGetByField(arity, index, value);
Assert.assertSame(value, tuple.getFloatByField(fieldName));
}
@@ -549,7 +585,8 @@ public class StormTupleTest extends AbstractTest {
Float value = r.nextFloat();
TestPojoMember<Float> pojo = new TestPojoMember<Float>(value);
StormTuple<TestPojoMember<Float>> tuple = new StormTuple<TestPojoMember<Float>>(pojo,
- null);
+ null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getFloatByField(fieldNamePojo));
}
@@ -558,16 +595,16 @@ public class StormTupleTest extends AbstractTest {
Float value = r.nextFloat();
TestPojoGetter<Float> pojo = new TestPojoGetter<Float>(value);
StormTuple<TestPojoGetter<Float>> tuple = new StormTuple<TestPojoGetter<Float>>(pojo,
- null);
+ null,
+ -1, null, null, null);
Assert.assertSame(value, tuple.getFloatByField(fieldNamePojo));
}
- @SuppressWarnings("rawtypes")
@Test
public void testGetBinaryByField() throws Exception {
byte[] data = new byte[1 + r.nextInt(20)];
r.nextBytes(data);
- StormTuple tuple = testGetByField(arity, index, data);
+ StormTuple<?> tuple = testGetByField(arity, index, data);
Assert.assertSame(data, tuple.getBinaryByField(fieldName));
}
@@ -577,7 +614,7 @@ public class StormTupleTest extends AbstractTest {
r.nextBytes(data);
TestPojoMember<byte[]> pojo = new TestPojoMember<byte[]>(data);
StormTuple<TestPojoMember<byte[]>> tuple = new StormTuple<TestPojoMember<byte[]>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(data, tuple.getBinaryByField(fieldNamePojo));
}
@@ -587,12 +624,11 @@ public class StormTupleTest extends AbstractTest {
r.nextBytes(data);
TestPojoGetter<byte[]> pojo = new TestPojoGetter<byte[]>(data);
StormTuple<TestPojoGetter<byte[]>> tuple = new StormTuple<TestPojoGetter<byte[]>>(pojo,
- null);
+ null, -1, null, null, null);
Assert.assertSame(data, tuple.getBinaryByField(fieldNamePojo));
}
- @SuppressWarnings({ "unchecked", "rawtypes" })
- private <T> StormTuple testGetByField(int arity, int index, T value)
+ private <T> StormTuple<?> testGetByField(int arity, int index, T value)
throws Exception {
assert (index < arity);
@@ -610,41 +646,42 @@ public class StormTupleTest extends AbstractTest {
}
Fields schema = new Fields(attributeNames);
- return new StormTuple(tuple, schema);
+ return new StormTuple<>(tuple, schema, -1, null, null, null);
}
@Test
public void testGetSourceGlobalStreamid() {
- GlobalStreamId globalStreamid =
- new StormTuple<>(new Tuple1<>(), null, 0, "streamId", "componentID").getSourceGlobalStreamid();
+ GlobalStreamId globalStreamid = new StormTuple<>(null, null, -1, "streamId", "componentID",
+ null).getSourceGlobalStreamid();
Assert.assertEquals("streamId", globalStreamid.get_streamId());
Assert.assertEquals("componentID", globalStreamid.get_componentId());
}
@Test
public void testGetSourceComponent() {
- String sourceComponent =
- new StormTuple<>(new Tuple1<>(), null, 0, "streamId", "componentID").getSourceComponent();
+ String sourceComponent = new StormTuple<>(null, null, -1, null, "componentID", null)
+ .getSourceComponent();
Assert.assertEquals("componentID", sourceComponent);
}
@Test
public void testGetSourceTask() {
- String sourceStreamId =
- new StormTuple<>(new Tuple1<>(), null, 0, "streamId", "componentID").getSourceStreamId();
- Assert.assertEquals("streamId", sourceStreamId);
+ int sourceTaskId = new StormTuple<>(null, null, 42, null, null, null).getSourceTask();
+ Assert.assertEquals(42, sourceTaskId);
}
@Test
public void testGetSourceStreamId() {
- String sourceStreamId =
- new StormTuple<>(new Tuple1<>(), null, 0, "streamId", "componentID").getSourceStreamId();
+ String sourceStreamId = new StormTuple<>(null, null, -1, "streamId", null, null)
+ .getSourceStreamId();
Assert.assertEquals("streamId", sourceStreamId);
}
@Test
public void testGetMessageId() {
- Assert.assertNotNull(new StormTuple<>(null, null).getMessageId());
+ MessageId messageId = MessageId.makeUnanchored();
+ StormTuple<?> stormTuple = new StormTuple<>(null, null, -1, null, null, messageId);
+ Assert.assertSame(messageId, stormTuple.getMessageId());
}
public static class TestPojoMember<T> {
[2/2] flink git commit: [FLINK-2721] [Storm Compatibility] Add Tuple
meta information
Posted by mj...@apache.org.
[FLINK-2721] [Storm Compatibility] Add Tuple meta information
This closes #1591
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7eb6edd5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7eb6edd5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7eb6edd5
Branch: refs/heads/master
Commit: 7eb6edd573211f6f65449ad5ed13eb588cebfe67
Parents: 168c1f0
Author: mjsax <mj...@informatik.hu-berlin.de>
Authored: Fri Feb 5 00:02:41 2016 +0100
Committer: mjsax <mj...@informatik.hu-berlin.de>
Committed: Sun Feb 7 19:39:08 2016 +0100
----------------------------------------------------------------------
.../flink/storm/tests/StormMetaDataITCase.java | 58 +++++
.../storm/tests/operators/MetaDataSpout.java | 55 +++++
.../tests/operators/VerifyMetaDataBolt.java | 69 ++++++
flink-contrib/flink-storm/README.md | 1 -
.../flink/storm/api/FlinkLocalCluster.java | 13 +-
.../storm/api/FlinkOutputFieldsDeclarer.java | 8 +-
.../apache/flink/storm/api/FlinkTopology.java | 21 +-
.../flink/storm/api/StormFlinkStreamMerger.java | 5 +-
.../flink/storm/api/TwoFlinkStreamsMerger.java | 7 +-
.../storm/wrappers/AbstractStormCollector.java | 49 +++-
.../flink/storm/wrappers/BoltCollector.java | 6 +-
.../flink/storm/wrappers/BoltWrapper.java | 65 ++++--
.../flink/storm/wrappers/SpoutCollector.java | 6 +-
.../flink/storm/wrappers/SpoutWrapper.java | 22 +-
.../apache/flink/storm/wrappers/StormTuple.java | 131 +++++++----
.../storm/wrappers/WrapperSetupHelper.java | 4 +-
.../api/FlinkOutputFieldsDeclarerTest.java | 18 +-
.../flink/storm/api/FlinkTopologyTest.java | 1 +
.../flink/storm/wrappers/BoltCollectorTest.java | 66 +++++-
.../flink/storm/wrappers/BoltWrapperTest.java | 50 +++-
.../storm/wrappers/SpoutCollectorTest.java | 71 +++++-
.../flink/storm/wrappers/StormTupleTest.java | 231 +++++++++++--------
22 files changed, 722 insertions(+), 235 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormMetaDataITCase.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormMetaDataITCase.java b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormMetaDataITCase.java
new file mode 100644
index 0000000..ce869df
--- /dev/null
+++ b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormMetaDataITCase.java
@@ -0,0 +1,58 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.flink.storm.tests;
+
+import backtype.storm.topology.TopologyBuilder;
+import backtype.storm.utils.Utils;
+
+import org.apache.flink.storm.api.FlinkLocalCluster;
+import org.apache.flink.storm.api.FlinkTopology;
+import org.apache.flink.storm.tests.operators.MetaDataSpout;
+import org.apache.flink.storm.tests.operators.VerifyMetaDataBolt;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.junit.Assert;
+
+public class StormMetaDataITCase extends StreamingProgramTestBase {
+
+ private final static String topologyId = "FieldsGrouping Test";
+ private final static String spoutId = "spout";
+ private final static String boltId1 = "bolt1";
+ private final static String boltId2 = "bolt2";
+
+ @Override
+ protected void testProgram() throws Exception {
+ final TopologyBuilder builder = new TopologyBuilder();
+
+ builder.setSpout(spoutId, new MetaDataSpout(), 2);
+ builder.setBolt(boltId1, new VerifyMetaDataBolt(), 2).localOrShuffleGrouping(spoutId,
+ MetaDataSpout.STREAM_ID);
+ builder.setBolt(boltId2, new VerifyMetaDataBolt()).shuffleGrouping(boltId1,
+ VerifyMetaDataBolt.STREAM_ID);
+
+ final FlinkLocalCluster cluster = FlinkLocalCluster.getLocalCluster();
+ cluster.submitTopology(topologyId, null, FlinkTopology.createTopology(builder));
+
+ // run topology for 5 seconds
+ Utils.sleep(5 * 1000);
+
+ cluster.shutdown();
+
+ Assert.assertFalse(VerifyMetaDataBolt.errorOccured);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/MetaDataSpout.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/MetaDataSpout.java b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/MetaDataSpout.java
new file mode 100644
index 0000000..2937909
--- /dev/null
+++ b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/MetaDataSpout.java
@@ -0,0 +1,55 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.flink.storm.tests.operators;
+
+import java.util.Map;
+
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseRichSpout;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+
+public class MetaDataSpout extends BaseRichSpout {
+ private static final long serialVersionUID = 5305870218033256376L;
+
+ public static final String STREAM_ID = "spoutMeta";
+
+ private SpoutOutputCollector collector;
+ private TopologyContext context;
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ this.collector = collector;
+ this.context = context;
+ }
+
+ @Override
+ public void nextTuple() {
+ this.collector.emit(STREAM_ID, new Values(this.context.getThisComponentId(), STREAM_ID,
+ this.context.getThisTaskId()));
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declareStream(STREAM_ID, new Fields("id", "sid", "tid"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/VerifyMetaDataBolt.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/VerifyMetaDataBolt.java b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/VerifyMetaDataBolt.java
new file mode 100644
index 0000000..a02f1f9
--- /dev/null
+++ b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/operators/VerifyMetaDataBolt.java
@@ -0,0 +1,69 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.flink.storm.tests.operators;
+
+import java.util.Map;
+
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseRichBolt;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+public class VerifyMetaDataBolt extends BaseRichBolt {
+ private static final long serialVersionUID = 1353222852073800478L;
+
+ public static final String STREAM_ID = "boltMeta";
+
+ private OutputCollector collector;
+ private TopologyContext context;
+
+ public static boolean errorOccured = false;
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ this.collector = collector;
+ this.context = context;
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ if (!input.getSourceComponent().equals(input.getString(0))
+ || !input.getSourceStreamId().equals(input.getString(1))
+ || !input.getSourceGlobalStreamid().get_componentId().equals(input.getString(0))
+ || !input.getSourceGlobalStreamid().get_streamId().equals(input.getString(1))
+ || input.getSourceTask() != input.getInteger(2).intValue()
+ || !input.getMessageId().equals(MessageId.makeUnanchored())
+ || input.getMessageId().getAnchors().size() != 0
+ || input.getMessageId().getAnchorsToIds().size() != 0) {
+ errorOccured = true;
+ }
+ this.collector.emit(STREAM_ID, new Values(this.context.getThisComponentId(), STREAM_ID,
+ this.context.getThisTaskId()));
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declareStream(STREAM_ID, new Fields("id", "sid", "tid"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/README.md
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/README.md b/flink-contrib/flink-storm/README.md
index fd0f455..578d6cc 100644
--- a/flink-contrib/flink-storm/README.md
+++ b/flink-contrib/flink-storm/README.md
@@ -8,7 +8,6 @@ The code that builds the topology itself, can be reused unmodified. See `flink-s
**Please note**: Do not add `storm-core` as a dependency. It is already included via `flink-storm`.
The following Storm features are not (yet/fully) supported by the compatibility layer right now:
-* tuple meta information
* no fault-tolerance guarantees (ie, calls to `ack()`/`fail()` and anchoring is ignored)
* for whole Storm topologies the following is not supported by Flink:
* direct emit connection pattern
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkLocalCluster.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkLocalCluster.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkLocalCluster.java
index 04374fd..da19a22 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkLocalCluster.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkLocalCluster.java
@@ -45,7 +45,8 @@ public class FlinkLocalCluster {
/** The log used by this mini cluster */
private static final Logger LOG = LoggerFactory.getLogger(FlinkLocalCluster.class);
- /** The flink mini cluster on which to execute the programs */
+
+ /** The Flink mini cluster on which to execute the programs */
private FlinkMiniCluster flink;
/** Configuration key to submit topology in blocking mode if flag is set to {@code true}. */
@@ -122,8 +123,10 @@ public class FlinkLocalCluster {
}
public void shutdown() {
- flink.stop();
- flink = null;
+ if (this.flink != null) {
+ this.flink.stop();
+ this.flink = null;
+ }
}
public String getTopologyConf(final String id) {
@@ -183,8 +186,8 @@ public class FlinkLocalCluster {
public static interface LocalClusterFactory {
/**
- * Creates a local flink cluster.
- * @return A local flink cluster.
+ * Creates a local Flink cluster.
+ * @return A local Flink cluster.
*/
FlinkLocalCluster createLocalCluster();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarer.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarer.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarer.java
index 7c8d1ec..794beee 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarer.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarer.java
@@ -104,20 +104,20 @@ final class FlinkOutputFieldsDeclarer implements OutputFieldsDeclarer {
Tuple t;
final int numberOfAttributes = outputSchema.size();
- if (numberOfAttributes <= 25) {
+ if (numberOfAttributes <= 24) {
try {
- t = Tuple.getTupleClass(numberOfAttributes).newInstance();
+ t = Tuple.getTupleClass(numberOfAttributes + 1).newInstance();
} catch (final InstantiationException e) {
throw new RuntimeException(e);
} catch (final IllegalAccessException e) {
throw new RuntimeException(e);
}
} else {
- throw new IllegalArgumentException("Flink supports only a maximum number of 25 attributes");
+ throw new IllegalArgumentException("Flink supports only a maximum number of 24 attributes");
}
// TODO: declare only key fields as DefaultComparable
- for (int i = 0; i < numberOfAttributes; ++i) {
+ for (int i = 0; i < numberOfAttributes + 1; ++i) {
t.setField(new DefaultComparable(), i);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java
index a273188..811cfb4 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java
@@ -229,9 +229,24 @@ public class FlinkTopology {
boolean makeProgress = true;
while (bolts.size() > 0) {
if (!makeProgress) {
- throw new RuntimeException(
- "Unable to build Topology. Could not connect the following bolts: "
- + bolts.keySet());
+ StringBuilder strBld = new StringBuilder();
+ strBld.append("Unable to build Topology. Could not connect the following bolts:");
+ for (String boltId : bolts.keySet()) {
+ strBld.append("\n ");
+ strBld.append(boltId);
+ strBld.append(": missing input streams [");
+ for (Entry<GlobalStreamId, Grouping> streams : unprocessdInputsPerBolt
+ .get(boltId)) {
+ strBld.append("'");
+ strBld.append(streams.getKey().get_streamId());
+ strBld.append("' from '");
+ strBld.append(streams.getKey().get_componentId());
+ strBld.append("'; ");
+ }
+ strBld.append("]");
+ }
+
+ throw new RuntimeException(strBld.toString());
}
makeProgress = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/StormFlinkStreamMerger.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/StormFlinkStreamMerger.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/StormFlinkStreamMerger.java
index 2e8d3ad..72c1569 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/StormFlinkStreamMerger.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/StormFlinkStreamMerger.java
@@ -18,6 +18,7 @@ import org.apache.flink.util.Collector;
import backtype.storm.generated.GlobalStreamId;
import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
/**
* Merges a stream of type {@link StormTuple} with a Flink {@link DataStreams} into a stream of type {@link StormTuple}.
@@ -44,6 +45,6 @@ final class StormFlinkStreamMerger<IN1, IN2> implements CoFlatMapFunction<StormT
@Override
public void flatMap2(IN2 value, Collector<StormTuple> out) throws Exception {
out.collect(new StormTuple<IN2>(value, this.inputSchema, 0, this.inputStreamId,
- this.inputComponentId));
+ this.inputComponentId, MessageId.makeUnanchored()));
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/TwoFlinkStreamsMerger.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/TwoFlinkStreamsMerger.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/TwoFlinkStreamsMerger.java
index fb14a3b..1866e32 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/TwoFlinkStreamsMerger.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/TwoFlinkStreamsMerger.java
@@ -18,6 +18,7 @@ import org.apache.flink.util.Collector;
import backtype.storm.generated.GlobalStreamId;
import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
/**
* Merges two Flink {@link DataStreams} into a stream of type {@link StormTuple}.
@@ -46,12 +47,12 @@ final class TwoFlinkStreamsMerger<IN1, IN2> implements CoFlatMapFunction<IN1, IN
@Override
public void flatMap1(IN1 value, Collector<StormTuple> out) throws Exception {
out.collect(new StormTuple<IN1>(value, this.inputSchema1, 0,
- this.inputStreamId1, this.inputComponentId1));
+ this.inputStreamId1, this.inputComponentId1, MessageId.makeUnanchored()));
}
@Override
public void flatMap2(IN2 value, Collector<StormTuple> out) throws Exception {
out.collect(new StormTuple<IN2>(value, this.inputSchema2, 0,
- this.inputStreamId2, this.inputComponentId2));
+ this.inputStreamId2, this.inputComponentId2, MessageId.makeUnanchored()));
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java
index 85d895c..7f4d7d1 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/AbstractStormCollector.java
@@ -41,6 +41,8 @@ abstract class AbstractStormCollector<OUT> {
protected final HashMap<String, Integer> numberOfAttributes;
/** Indicates of multiple output stream are declared and thus {@link SplitStreamType} must be used as output. */
private final boolean split;
+ /** The ID of the producer task. */
+ private final int taskId;
/** Is set to {@code true} each time a tuple is emitted. */
boolean tupleEmitted = false;
@@ -51,25 +53,42 @@ abstract class AbstractStormCollector<OUT> {
*
* @param numberOfAttributes
* The number of attributes of the emitted tuples per output stream.
+ * @param taskId
+ * The ID of the producer task (negative value for unknown).
* @throws UnsupportedOperationException
- * if the specified number of attributes is greater than 25
+ * if the specified number of attributes is greater than 25 or taskId support is enabled for a raw
+ * stream
*/
- AbstractStormCollector(final HashMap<String, Integer> numberOfAttributes)
+ AbstractStormCollector(final HashMap<String, Integer> numberOfAttributes, final int taskId)
throws UnsupportedOperationException {
assert (numberOfAttributes != null);
this.numberOfAttributes = numberOfAttributes;
this.split = this.numberOfAttributes.size() > 1;
+ this.taskId = taskId;
for (Entry<String, Integer> outputStream : numberOfAttributes.entrySet()) {
- final int numAtt = outputStream.getValue();
- assert (numAtt >= -1);
+ int numAtt = outputStream.getValue();
+
+ if (this.taskId >= 0) {
+ if (numAtt < 0) {
+ throw new UnsupportedOperationException(
+ "Task ID transmission not supported for raw streams: "
+ + outputStream.getKey());
+ }
+ ++numAtt;
+ }
if (numAtt > 25) {
- throw new UnsupportedOperationException(
- "Flink cannot handle more then 25 attributes, but " + numAtt
- + " are declared for stream '" + outputStream.getKey()
- + "' by the given bolt");
+ if (this.taskId >= 0) {
+ throw new UnsupportedOperationException(
+ "Flink cannot handle more then 25 attributes, but 25 (24 plus 1 for produer task ID) "
+ + " are declared for stream '" + outputStream.getKey() + "' by the given bolt.");
+ } else {
+ throw new UnsupportedOperationException(
+ "Flink cannot handle more then 25 attributes, but " + numAtt
+ + " are declared for stream '" + outputStream.getKey() + "' by the given bolt.");
+ }
} else if (numAtt >= 0) {
try {
this.outputTuple.put(outputStream.getKey(),
@@ -99,13 +118,21 @@ abstract class AbstractStormCollector<OUT> {
protected final List<Integer> tansformAndEmit(final String streamId, final List<Object> tuple) {
List<Integer> taskIds;
- final int numAtt = this.numberOfAttributes.get(streamId);
- if (numAtt > -1) {
+ int numAtt = this.numberOfAttributes.get(streamId);
+ int taskIdIdx = numAtt;
+ if (this.taskId >= 0 && numAtt < 0) {
+ numAtt = 1;
+ taskIdIdx = 0;
+ }
+ if (numAtt >= 0) {
assert (tuple.size() == numAtt);
Tuple out = this.outputTuple.get(streamId);
for (int i = 0; i < numAtt; ++i) {
out.setField(tuple.get(i), i);
}
+ if (this.taskId >= 0) {
+ out.setField(this.taskId, taskIdIdx);
+ }
if (this.split) {
this.splitTuple.streamId = streamId;
this.splitTuple.value = out;
@@ -117,7 +144,7 @@ abstract class AbstractStormCollector<OUT> {
} else {
assert (tuple.size() == 1);
- if (split) {
+ if (this.split) {
this.splitTuple.streamId = streamId;
this.splitTuple.value = tuple.get(0);
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java
index 58fd098..2196a1c 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltCollector.java
@@ -46,14 +46,16 @@ class BoltCollector<OUT> extends AbstractStormCollector<OUT> implements IOutputC
*
* @param numberOfAttributes
* The number of attributes of the emitted tuples per output stream.
+ * @param taskId
+ * The ID of the producer task (negative value for unknown).
* @param flinkOutput
* The Flink output object to be used.
* @throws UnsupportedOperationException
* if the specified number of attributes is greater than 25
*/
- BoltCollector(final HashMap<String, Integer> numberOfAttributes,
+ BoltCollector(final HashMap<String, Integer> numberOfAttributes, final int taskId,
final Collector<OUT> flinkOutput) throws UnsupportedOperationException {
- super(numberOfAttributes);
+ super(numberOfAttributes, taskId);
assert (flinkOutput != null);
this.flinkOutput = flinkOutput;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
index 0c0eda9..ef09a3a 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
@@ -16,14 +16,20 @@
*/
package org.apache.flink.storm.wrappers;
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.generated.Grouping;
import backtype.storm.generated.StormTopology;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
import backtype.storm.utils.Utils;
+
import com.google.common.collect.Sets;
+
import org.apache.flink.api.common.ExecutionConfig.GlobalJobParameters;
+import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple0;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple25;
@@ -36,6 +42,7 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import java.util.Collection;
import java.util.HashMap;
+import java.util.Map;
/**
* A {@link BoltWrapper} wraps an {@link IRichBolt} in order to execute the Storm bolt within a Flink Streaming program.
@@ -66,12 +73,12 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
/** The topology context of the bolt. */
private transient TopologyContext topologyContext;
- /** The stream ID of the input stream for this bolt. */
- private final String inputStreamId;
- /** The component ID of the input stream for this bolt. */
- private final String inputComponentId;
- /** The schema (ie, ordered field names) of the input stream. */
- private final Fields inputSchema;
+ /** The IDs of the input streams for this bolt per producer task ID. */
+ private final HashMap<Integer, String> inputStreamIds = new HashMap<Integer, String>();
+ /** The IDs of the producres for this bolt per producer task ID.. */
+ private final HashMap<Integer, String> inputComponentIds = new HashMap<Integer, String>();
+ /** The schema (ie, ordered field names) of the input streams per producer taskID. */
+ private final HashMap<Integer, Fields> inputSchemas = new HashMap<Integer, Fields>();
/**
* We have to use this because Operators must output {@link StreamRecord}.
@@ -216,12 +223,6 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
* The Storm {@link IRichBolt bolt} to be used.
* @param name
* The name of the bolt.
- * @param inputStreamId
- * The stream id of the input stream for this bolt
- * @param inputComponentId
- * The component id of the input stream for this bolt
- * @param inputSchema
- * The schema (ie, ordered field names) of the input stream.
* @param rawOutputs
* Contains stream names if a single attribute output stream, should not be of type {@link Tuple1} but be
* of a raw type.
@@ -235,9 +236,7 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
final Collection<String> rawOutputs) throws IllegalArgumentException {
this.bolt = bolt;
this.name = name;
- this.inputComponentId = inputComponentId;
- this.inputStreamId = inputStreamId;
- this.inputSchema = inputSchema;
+ this.inputSchemas.put(null, inputSchema);
this.numberOfAttributes = WrapperSetupHelper.getNumberOfAttributes(bolt, rawOutputs);
}
@@ -256,8 +255,6 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
super.open();
this.flinkCollector = new TimestampedCollector<>(this.output);
- final OutputCollector stormCollector = new OutputCollector(new BoltCollector<OUT>(
- this.numberOfAttributes, this.flinkCollector));
GlobalJobParameters config = getExecutionConfig().getGlobalJobParameters();
StormConfig stormConfig = new StormConfig();
@@ -272,6 +269,24 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
this.topologyContext = WrapperSetupHelper.createTopologyContext(
getRuntimeContext(), this.bolt, this.name, this.stormTopology, stormConfig);
+
+ final OutputCollector stormCollector = new OutputCollector(new BoltCollector<OUT>(
+ this.numberOfAttributes, this.topologyContext.getThisTaskId(), this.flinkCollector));
+
+ if (this.stormTopology != null) {
+ Map<GlobalStreamId, Grouping> inputs = this.topologyContext.getThisSources();
+
+ for (GlobalStreamId inputStream : inputs.keySet()) {
+ for (Integer tid : this.topologyContext.getComponentTasks(inputStream
+ .get_componentId())) {
+ this.inputComponentIds.put(tid, inputStream.get_componentId());
+ this.inputStreamIds.put(tid, inputStream.get_streamId());
+ this.inputSchemas.put(tid,
+ this.topologyContext.getComponentOutputFields(inputStream));
+ }
+ }
+ }
+
this.bolt.prepare(stormConfig, this.topologyContext, stormCollector);
}
@@ -283,9 +298,21 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
@Override
public void processElement(final StreamRecord<IN> element) throws Exception {
this.flinkCollector.setTimestamp(element.getTimestamp());
+
IN value = element.getValue();
- this.bolt.execute(new StormTuple<>(value, this.inputSchema, this.topologyContext
- .getThisTaskId(), this.inputStreamId, this.inputComponentId));
+
+ if (this.stormTopology != null) {
+ Tuple tuple = (Tuple) value;
+ Integer producerTaskId = tuple.getField(tuple.getArity() - 1);
+
+ this.bolt.execute(new StormTuple<>(value, this.inputSchemas.get(producerTaskId),
+ producerTaskId, this.inputStreamIds.get(producerTaskId), this.inputComponentIds
+ .get(producerTaskId), MessageId.makeUnanchored()));
+
+ } else {
+ this.bolt.execute(new StormTuple<>(value, this.inputSchemas.get(null), -1, null, null,
+ MessageId.makeUnanchored()));
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java
index 4e89844..0e2190e 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutCollector.java
@@ -42,14 +42,16 @@ class SpoutCollector<OUT> extends AbstractStormCollector<OUT> implements ISpoutO
*
* @param numberOfAttributes
* The number of attributes of the emitted tuples.
+ * @param taskId
+ * The ID of the producer task (negative value for unknown).
* @param flinkContext
* The Flink source context to be used.
* @throws UnsupportedOperationException
* if the specified number of attributes is greater than 25
*/
- SpoutCollector(final HashMap<String, Integer> numberOfAttributes,
+ SpoutCollector(final HashMap<String, Integer> numberOfAttributes, final int taskId,
final SourceContext<OUT> flinkContext) throws UnsupportedOperationException {
- super(numberOfAttributes);
+ super(numberOfAttributes, taskId);
assert (flinkContext != null);
this.flinkContext = flinkContext;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java
index 482e7c5..e077aeb 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java
@@ -19,8 +19,11 @@ package org.apache.flink.storm.wrappers;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
+
import com.google.common.collect.Sets;
+
import org.apache.flink.api.common.ExecutionConfig.GlobalJobParameters;
import org.apache.flink.api.java.tuple.Tuple0;
import org.apache.flink.api.java.tuple.Tuple1;
@@ -58,8 +61,6 @@ public final class SpoutWrapper<OUT> extends RichParallelSourceFunction<OUT> {
private final IRichSpout spout;
/** The name of the spout. */
private final String name;
- /** The wrapper of the given Flink collector. */
- private SpoutCollector<OUT> collector;
/** Indicates, if the source is still running or was canceled. */
private volatile boolean isRunning = true;
/** The number of {@link IRichSpout#nextTuple()} calls. */
@@ -239,9 +240,7 @@ public final class SpoutWrapper<OUT> extends RichParallelSourceFunction<OUT> {
@Override
public final void run(final SourceContext<OUT> ctx) throws Exception {
- this.collector = new SpoutCollector<OUT>(this.numberOfAttributes, ctx);
-
- GlobalJobParameters config = super.getRuntimeContext().getExecutionConfig()
+ final GlobalJobParameters config = super.getRuntimeContext().getExecutionConfig()
.getGlobalJobParameters();
StormConfig stormConfig = new StormConfig();
@@ -253,9 +252,14 @@ public final class SpoutWrapper<OUT> extends RichParallelSourceFunction<OUT> {
}
}
- this.spout.open(stormConfig, WrapperSetupHelper.createTopologyContext(
+ final TopologyContext stormTopologyContext = WrapperSetupHelper.createTopologyContext(
(StreamingRuntimeContext) super.getRuntimeContext(), this.spout, this.name,
- this.stormTopology, stormConfig), new SpoutOutputCollector(this.collector));
+ this.stormTopology, stormConfig);
+
+ SpoutCollector<OUT> collector = new SpoutCollector<OUT>(this.numberOfAttributes,
+ stormTopologyContext.getThisTaskId(), ctx);
+
+ this.spout.open(stormConfig, stormTopologyContext, new SpoutOutputCollector(collector));
this.spout.activate();
if (numberOfInvocations == null) {
@@ -278,9 +282,9 @@ public final class SpoutWrapper<OUT> extends RichParallelSourceFunction<OUT> {
}
} else {
do {
- this.collector.tupleEmitted = false;
+ collector.tupleEmitted = false;
this.spout.nextTuple();
- } while (this.collector.tupleEmitted && this.isRunning);
+ } while (collector.tupleEmitted && this.isRunning);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java
index e4c0805..febf0f3 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/StormTuple.java
@@ -29,8 +29,6 @@ import backtype.storm.generated.GlobalStreamId;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.MessageId;
import backtype.storm.tuple.Values;
-import backtype.storm.utils.Utils;
-import com.google.common.base.Preconditions;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
@@ -41,55 +39,61 @@ import java.util.List;
*/
public class StormTuple<IN> implements backtype.storm.tuple.Tuple {
- /** The Storm representation of the original Flink tuple */
+ /** The Storm representation of the original Flink tuple. */
private final Values stormTuple;
- /** The schema (ie, ordered field names) of the tuple */
+ /** The schema (ie, ordered field names) of this tuple. */
private final Fields schema;
-
- /** The task id where this tuple is processed */
- private final int taskId;
- /** The producer of this tuple */
+ /** The task ID where this tuple was produced. */
+ private final int producerTaskId;
+ /** The input stream from which this tuple was received. */
private final String producerStreamId;
- /** The producer's component id of this tuple */
+ /** The producer's component ID of this tuple. */
private final String producerComponentId;
- /** The message that is associated with this tuple */
- private final MessageId id;
+ /** The message that is associated with this tuple. */
+ private final MessageId messageId;
- /**
- * Constructor which sets defaults for producerComponentId, taskId, and componentID
- * @param flinkTuple the Flink tuple
- * @param schema The schema of the storm fields
- */
- public StormTuple(final IN flinkTuple, final Fields schema) {
- this(flinkTuple, schema, -1, Utils.DEFAULT_STREAM_ID, BoltWrapper.DEFAULT_ID);
- }
/**
- * Create a new Storm tuple from the given Flink tuple. The provided {@code nameIndexMap} is ignored for raw input
- * types.
- * @param flinkTuple The Flink tuple to be converted.
- * @param schema The schema (ie, ordered field names) of the tuple.
- * @param producerComponentId The component id of the producer.
+ * Create a new Storm tuple from the given Flink tuple.
+ *
+ * @param flinkTuple
+ * The Flink tuple to be converted.
+ * @param schema
+ * The schema (ie, ordered field names) of the tuple.
+ * @param producerTaskId
+ * The task ID of the producer (a valid, ie, non-negative ID, implicates the truncation of the last
+ * attribute of {@code flinkTuple}).
+ * @param producerStreamId
+ * The input stream ID from which this tuple was received.
+ * @param producerComponentId
+ * The component ID of the producer.
+ * @param messageId
+ * The message ID of this tuple.
*/
- public StormTuple(final IN flinkTuple, final Fields schema, int taskId, String producerStreamId, String producerComponentId) {
+ public StormTuple(final IN flinkTuple, final Fields schema, final int producerTaskId, final String producerStreamId, final String producerComponentId, final MessageId messageId) {
if (flinkTuple instanceof org.apache.flink.api.java.tuple.Tuple) {
- this.schema = schema;
final org.apache.flink.api.java.tuple.Tuple t = (org.apache.flink.api.java.tuple.Tuple) flinkTuple;
- final int numberOfAttributes = t.getArity();
+ final int numberOfAttributes;
+ // does flinkTuple carry producerTaskId as last attribute?
+ if (producerTaskId < 0) {
+ numberOfAttributes = t.getArity();
+ } else {
+ numberOfAttributes = t.getArity() - 1;
+ }
this.stormTuple = new Values();
for (int i = 0; i < numberOfAttributes; ++i) {
this.stormTuple.add(t.getField(i));
}
} else {
- this.schema = null;
this.stormTuple = new Values(flinkTuple);
}
- this.taskId = Preconditions.checkNotNull(taskId);
- this.producerStreamId = Preconditions.checkNotNull(producerStreamId);
- this.producerComponentId = Preconditions.checkNotNull(producerComponentId);
- this.id = Preconditions.checkNotNull(MessageId.makeUnanchored());
+ this.schema = schema;
+ this.producerTaskId = producerTaskId;
+ this.producerStreamId = producerStreamId;
+ this.producerComponentId = producerComponentId;
+ this.messageId = messageId;
}
@Override
@@ -289,54 +293,92 @@ public class StormTuple<IN> implements backtype.storm.tuple.Tuple {
@Override
public GlobalStreamId getSourceGlobalStreamid() {
- return new GlobalStreamId(getSourceComponent(), producerStreamId);
+ return new GlobalStreamId(this.producerComponentId, this.producerStreamId);
}
@Override
public String getSourceComponent() {
- return producerComponentId;
+ return this.producerComponentId;
}
@Override
public int getSourceTask() {
- return taskId;
+ return this.producerTaskId;
}
@Override
public String getSourceStreamId() {
- return producerStreamId;
+ return this.producerStreamId;
}
@Override
public MessageId getMessageId() {
- return id;
+ return this.messageId;
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
- result = (prime * result) + ((this.stormTuple == null) ? 0 : this.stormTuple.hashCode());
+ result = prime * result + ((messageId == null) ? 0 : messageId.hashCode());
+ result = prime * result
+ + ((producerComponentId == null) ? 0 : producerComponentId.hashCode());
+ result = prime * result + ((producerStreamId == null) ? 0 : producerStreamId.hashCode());
+ result = prime * result + producerTaskId;
+ result = prime * result + ((schema == null) ? 0 : schema.toList().hashCode());
+ result = prime * result + ((stormTuple == null) ? 0 : stormTuple.hashCode());
return result;
}
+ @SuppressWarnings("rawtypes")
@Override
- public boolean equals(final Object obj) {
+ public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null) {
return false;
}
- if (this.getClass() != obj.getClass()) {
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ StormTuple other = (StormTuple) obj;
+ if (messageId == null) {
+ if (other.messageId != null) {
+ return false;
+ }
+ } else if (!messageId.equals(other.messageId)) {
+ return false;
+ }
+ if (producerComponentId == null) {
+ if (other.producerComponentId != null) {
+ return false;
+ }
+ } else if (!producerComponentId.equals(other.producerComponentId)) {
+ return false;
+ }
+ if (producerStreamId == null) {
+ if (other.producerStreamId != null) {
+ return false;
+ }
+ } else if (!producerStreamId.equals(other.producerStreamId)) {
+ return false;
+ }
+ if (producerTaskId != other.producerTaskId) {
+ return false;
+ }
+ if (schema == null) {
+ if (other.schema != null) {
+ return false;
+ }
+ } else if (!schema.toList().equals(other.schema.toList())) {
return false;
}
- final StormTuple<?> other = (StormTuple<?>) obj;
- if (this.stormTuple == null) {
+ if (stormTuple == null) {
if (other.stormTuple != null) {
return false;
}
- } else if (!this.stormTuple.equals(other.stormTuple)) {
+ } else if (!stormTuple.equals(other.stormTuple)) {
return false;
}
return true;
@@ -344,6 +386,7 @@ public class StormTuple<IN> implements backtype.storm.tuple.Tuple {
@Override
public String toString() {
- return "StormTuple{ "+ stormTuple.toString() +" }";
+ return "StormTuple{ " + stormTuple.toString() + "[" + this.producerComponentId + ","
+ + this.producerStreamId + "," + this.producerTaskId + "," + this.messageId + "]}";
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java
index e384593..74a12dd 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/WrapperSetupHelper.java
@@ -120,7 +120,7 @@ class WrapperSetupHelper {
String stormId = (String) stormConfig.get(TOPOLOGY_NAME);
String codeDir = null; // not supported
String pidDir = null; // not supported
- Integer taskId = null;
+ Integer taskId = -1;
Integer workerPort = null; // not supported
List<Integer> workerTasks = new ArrayList<Integer>();
final Map<String, Object> defaultResources = new HashMap<String, Object>();
@@ -144,8 +144,6 @@ class WrapperSetupHelper {
}
stormTopology = new StormTopology(spouts, bolts, new HashMap<String, StateSpoutSpec>());
- taskId = context.getIndexOfThisSubtask();
-
List<Integer> sortedTasks = new ArrayList<Integer>(dop);
for (int i = 1; i <= dop; ++i) {
taskToComponents.put(i, operatorName);
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java
index 6c41f17..90a82ba 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkOutputFieldsDeclarerTest.java
@@ -36,7 +36,7 @@ public class FlinkOutputFieldsDeclarerTest extends AbstractTest {
public void testDeclare() {
for (int i = 0; i < 2; ++i) { // test case: simple / non-direct
for (int j = 1; j < 2; ++j) { // number of streams
- for (int k = 0; k <= 25; ++k) { // number of attributes
+ for (int k = 0; k <= 24; ++k) { // number of attributes
this.runDeclareTest(i, j, k);
}
}
@@ -45,22 +45,22 @@ public class FlinkOutputFieldsDeclarerTest extends AbstractTest {
@Test(expected = IllegalArgumentException.class)
public void testDeclareSimpleToManyAttributes() {
- this.runDeclareTest(0, this.r.nextBoolean() ? 1 : 2, 26);
+ this.runDeclareTest(0, this.r.nextBoolean() ? 1 : 2, 25);
}
@Test(expected = IllegalArgumentException.class)
public void testDeclareNonDirectToManyAttributes() {
- this.runDeclareTest(1, this.r.nextBoolean() ? 1 : 2, 26);
+ this.runDeclareTest(1, this.r.nextBoolean() ? 1 : 2, 25);
}
@Test(expected = IllegalArgumentException.class)
public void testDeclareDefaultStreamToManyAttributes() {
- this.runDeclareTest(2, this.r.nextBoolean() ? 1 : 2, 26);
+ this.runDeclareTest(2, this.r.nextBoolean() ? 1 : 2, 25);
}
@Test(expected = IllegalArgumentException.class)
public void testDeclareFullToManyAttributes() {
- this.runDeclareTest(3, this.r.nextBoolean() ? 1 : 2, 26);
+ this.runDeclareTest(3, this.r.nextBoolean() ? 1 : 2, 25);
}
private void runDeclareTest(final int testCase, final int numberOfStreams,
@@ -95,7 +95,7 @@ public class FlinkOutputFieldsDeclarerTest extends AbstractTest {
for (String stream : streams) {
final TypeInformation<?> type = declarer.getOutputType(stream);
- Assert.assertEquals(numberOfAttributes, type.getArity());
+ Assert.assertEquals(numberOfAttributes + 1, type.getArity());
Assert.assertTrue(type.isTupleType());
}
}
@@ -142,7 +142,7 @@ public class FlinkOutputFieldsDeclarerTest extends AbstractTest {
@Test
public void testGetGroupingFieldIndexes() {
- final int numberOfAttributes = 5 + this.r.nextInt(21);
+ final int numberOfAttributes = 5 + this.r.nextInt(20);
final String[] attributes = new String[numberOfAttributes];
for (int i = 0; i < numberOfAttributes; ++i) {
attributes[i] = "a" + i;
@@ -151,12 +151,12 @@ public class FlinkOutputFieldsDeclarerTest extends AbstractTest {
final FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer();
declarer.declare(new Fields(attributes));
- final int numberOfKeys = 1 + this.r.nextInt(25);
+ final int numberOfKeys = 1 + this.r.nextInt(24);
final LinkedList<String> groupingFields = new LinkedList<String>();
final boolean[] indexes = new boolean[numberOfAttributes];
for (int i = 0; i < numberOfAttributes; ++i) {
- if (this.r.nextInt(26) < numberOfKeys) {
+ if (this.r.nextInt(25) < numberOfKeys) {
groupingFields.add(attributes[i]);
indexes[i] = true;
} else {
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java
index 5526b2e..39b01d8 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/api/FlinkTopologyTest.java
@@ -21,6 +21,7 @@ import backtype.storm.tuple.Fields;
import org.apache.flink.storm.util.TestDummyBolt;
import org.apache.flink.storm.util.TestDummySpout;
import org.apache.flink.storm.util.TestSink;
+
import org.junit.Assert;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltCollectorTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltCollectorTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltCollectorTest.java
index 8124161..e8748d0 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltCollectorTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltCollectorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.storm.wrappers;
import backtype.storm.tuple.Values;
+
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.storm.util.AbstractTest;
import org.apache.flink.streaming.api.operators.Output;
@@ -48,11 +49,11 @@ public class BoltCollectorTest extends AbstractTest {
attributes.put(streamId, numberOfAttributes);
if (numberOfAttributes == -1) {
- collector = new BoltCollector(attributes, flinkCollector);
+ collector = new BoltCollector(attributes, -1, flinkCollector);
tuple.add(new Integer(this.r.nextInt()));
} else {
- collector = new BoltCollector(attributes, flinkCollector);
+ collector = new BoltCollector(attributes, -1, flinkCollector);
flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance();
for (int i = 0; i < numberOfAttributes; ++i) {
@@ -75,10 +76,69 @@ public class BoltCollectorTest extends AbstractTest {
}
}
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testBoltStormCollectorWithTaskId() throws InstantiationException, IllegalAccessException {
+ for (int numberOfAttributes = 0; numberOfAttributes < 25; ++numberOfAttributes) {
+ final Output flinkCollector = mock(Output.class);
+ final int taskId = 42;
+ final String streamId = "streamId";
+
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put(streamId, numberOfAttributes);
+
+ BoltCollector<?> collector = new BoltCollector(attributes, taskId, flinkCollector);
+
+ final Values tuple = new Values();
+ final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes + 1).newInstance();
+
+ for (int i = 0; i < numberOfAttributes; ++i) {
+ tuple.add(new Integer(this.r.nextInt()));
+ flinkTuple.setField(tuple.get(i), i);
+ }
+ flinkTuple.setField(taskId, numberOfAttributes);
+
+ final Collection anchors = mock(Collection.class);
+ final List<Integer> taskIds;
+ taskIds = collector.emit(streamId, anchors, tuple);
+
+ Assert.assertNull(taskIds);
+
+ verify(flinkCollector).collect(flinkTuple);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testToManyAttributes() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", 26);
+
+ new BoltCollector<Object>(attributes, -1, mock(Output.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testToManyAttributesWithTaskId() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", 25);
+
+ new BoltCollector<Object>(attributes, 42, mock(Output.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRawStreamWithTaskId() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", -1);
+
+ new BoltCollector<Object>(attributes, 42, mock(Output.class));
+ }
+
@SuppressWarnings("unchecked")
@Test(expected = UnsupportedOperationException.class)
public void testEmitDirect() {
- new BoltCollector<Object>(mock(HashMap.class), mock(Output.class)).emitDirect(0, null,
+ new BoltCollector<Object>(mock(HashMap.class), -1, mock(Output.class)).emitDirect(0, null,
null, null);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
index 69e967d..1a583d4 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
@@ -22,8 +22,10 @@ import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
+
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.java.tuple.Tuple;
@@ -152,9 +154,13 @@ public class BoltWrapperTest extends AbstractTest {
wrapper.processElement(record);
if (numberOfAttributes == -1) {
- verify(bolt).execute(eq(new StormTuple<String>(rawTuple, null)));
+ verify(bolt).execute(
+ eq(new StormTuple<String>(rawTuple, null, -1, null, null, MessageId
+ .makeUnanchored())));
} else {
- verify(bolt).execute(eq(new StormTuple<Tuple>(flinkTuple, null)));
+ verify(bolt).execute(
+ eq(new StormTuple<Tuple>(flinkTuple, null, -1, null, null, MessageId
+ .makeUnanchored())));
}
}
@@ -207,12 +213,23 @@ public class BoltWrapperTest extends AbstractTest {
@SuppressWarnings("unchecked")
@Test
public void testOpen() throws Exception {
-
+
// utility mocks
+ final StormConfig stormConfig = new StormConfig();
+ final Configuration flinkConfig = new Configuration();
+
+ final ExecutionConfig taskConfig = mock(ExecutionConfig.class);
+ when(taskConfig.getGlobalJobParameters()).thenReturn(null).thenReturn(stormConfig)
+ .thenReturn(flinkConfig);
+
+ final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
+ when(taskContext.getExecutionConfig()).thenReturn(taskConfig);
+ when(taskContext.getTaskName()).thenReturn("name");
+
final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
declarer.declare(new Fields("dummy"));
PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);
-
+
// (1) open with no configuration
{
ExecutionConfig execConfig = mock(ExecutionConfig.class);
@@ -228,8 +245,6 @@ public class BoltWrapperTest extends AbstractTest {
// (2) open with a storm specific configuration
{
- final StormConfig stormConfig = new StormConfig();
-
ExecutionConfig execConfig = mock(ExecutionConfig.class);
when(execConfig.getGlobalJobParameters()).thenReturn(stormConfig);
@@ -264,12 +279,23 @@ public class BoltWrapperTest extends AbstractTest {
@SuppressWarnings("unchecked")
@Test
public void testOpenSink() throws Exception {
+ final StormConfig stormConfig = new StormConfig();
+ final Configuration flinkConfig = new Configuration();
+
+ final ExecutionConfig taskConfig = mock(ExecutionConfig.class);
+ when(taskConfig.getGlobalJobParameters()).thenReturn(null).thenReturn(stormConfig)
+ .thenReturn(flinkConfig);
+
+ final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
+ when(taskContext.getExecutionConfig()).thenReturn(taskConfig);
+ when(taskContext.getTaskName()).thenReturn("name");
+
final IRichBolt bolt = mock(IRichBolt.class);
BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
-
+
wrapper.setup(createMockStreamTask(), new StreamConfig(new Configuration()), mock(Output.class));
wrapper.open();
-
+
verify(bolt).prepare(any(Map.class), any(TopologyContext.class), isNotNull(OutputCollector.class));
}
@@ -283,7 +309,7 @@ public class BoltWrapperTest extends AbstractTest {
PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);
final BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
-
+
wrapper.setup(createMockStreamTask(), new StreamConfig(new Configuration()), mock(Output.class));
wrapper.close();
@@ -330,18 +356,18 @@ public class BoltWrapperTest extends AbstractTest {
public static StreamTask<?, ?> createMockStreamTask() {
return createMockStreamTask(new ExecutionConfig());
}
-
+
public static StreamTask<?, ?> createMockStreamTask(ExecutionConfig execConfig) {
Environment env = mock(Environment.class);
when(env.getTaskInfo()).thenReturn(new TaskInfo("Mock Task", 0, 1, 0));
when(env.getUserClassLoader()).thenReturn(BoltWrapperTest.class.getClassLoader());
-
+
StreamTask<?, ?> mockTask = mock(StreamTask.class);
when(mockTask.getCheckpointLock()).thenReturn(new Object());
when(mockTask.getConfiguration()).thenReturn(new StreamConfig(new Configuration()));
when(mockTask.getEnvironment()).thenReturn(env);
when(mockTask.getExecutionConfig()).thenReturn(execConfig);
-
+
return mockTask;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7eb6edd5/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java
index b3a0170..fac2582 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutCollectorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.storm.wrappers;
import backtype.storm.tuple.Values;
+
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.storm.util.AbstractTest;
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
@@ -47,11 +48,10 @@ public class SpoutCollectorTest extends AbstractTest {
attributes.put(streamId, numberOfAttributes);
if (numberOfAttributes == -1) {
- collector = new SpoutCollector(attributes, flinkCollector);
+ collector = new SpoutCollector(attributes, -1, flinkCollector);
tuple.add(new Integer(this.r.nextInt()));
-
} else {
- collector = new SpoutCollector(attributes, flinkCollector);
+ collector = new SpoutCollector(attributes, -1, flinkCollector);
flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance();
for (int i = 0; i < numberOfAttributes; ++i) {
@@ -75,12 +75,71 @@ public class SpoutCollectorTest extends AbstractTest {
}
}
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testSpoutStormCollectorWithTaskId() throws InstantiationException, IllegalAccessException {
+ for (int numberOfAttributes = 0; numberOfAttributes < 25; ++numberOfAttributes) {
+ final SourceContext flinkCollector = mock(SourceContext.class);
+ final int taskId = 42;
+ final String streamId = "streamId";
+
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put(streamId, numberOfAttributes);
+
+ SpoutCollector<?> collector = new SpoutCollector(attributes, taskId, flinkCollector);
+
+ final Values tuple = new Values();
+ final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes + 1).newInstance();
+
+ for (int i = 0; i < numberOfAttributes; ++i) {
+ tuple.add(new Integer(this.r.nextInt()));
+ flinkTuple.setField(tuple.get(i), i);
+ }
+ flinkTuple.setField(taskId, numberOfAttributes);
+
+ final List<Integer> taskIds;
+ final Object messageId = new Integer(this.r.nextInt());
+
+ taskIds = collector.emit(streamId, tuple, messageId);
+
+ Assert.assertNull(taskIds);
+
+ verify(flinkCollector).collect(flinkTuple);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testTooManyAttributes() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", 26);
+
+ new SpoutCollector<Object>(attributes, -1, mock(SourceContext.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testTooManyAttributesWithTaskId() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", 25);
+
+ new SpoutCollector<Object>(attributes, 42, mock(SourceContext.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRawStreamWithTaskId() {
+ HashMap<String, Integer> attributes = new HashMap<String, Integer>();
+ attributes.put("", -1);
+
+ new SpoutCollector<Object>(attributes, 42, mock(SourceContext.class));
+ }
+
@SuppressWarnings("unchecked")
@Test(expected = UnsupportedOperationException.class)
public void testEmitDirect() {
- new SpoutCollector<Object>(mock(HashMap.class), mock(SourceContext.class)).emitDirect(
- 0, null, null,
- (Object) null);
+ new SpoutCollector<Object>(mock(HashMap.class), -1, mock(SourceContext.class)).emitDirect(
+ 0, null, null, (Object) null);
}
}