flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
/**
	 * Test Flink's Kafka integration also with very big records (30MB).
	 *
	 * <p>see http://stackoverflow.com/questions/21020347/kafka-sending-a-15mb-message
	 *
	 */
Test Flink's Kafka integration also with very big records (30MB).
Test Flink ' s Kafka integration also with very big records ( 30MB ) .
public void runBigRecordTestTopology() throws Exception {
		final String topic = "bigRecordTestTopic";
		final int parallelism = 1; 
		createTestTopic(topic, parallelism, 1);
		final TypeInformation<Tuple2<Long, byte[]>> longBytesInfo = TypeInfoParser.parse("Tuple2<Long, byte[]>");
		final TypeInformationSerializationSchema<Tuple2<Long, byte[]>> serSchema =
				new TypeInformationSerializationSchema<>(longBytesInfo, new ExecutionConfig());
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setRestartStrategy(RestartStrategies.noRestart());
		env.getConfig().disableSysoutLogging();
		env.enableCheckpointing(100);
		env.setParallelism(parallelism);
		Properties consumerProps = new Properties();
		consumerProps.putAll(standardProps);
		consumerProps.setProperty("fetch.message.max.bytes", Integer.toString(1024 * 1024 * 14));
		consumerProps.setProperty("max.partition.fetch.bytes", Integer.toString(1024 * 1024 * 14)); 
		consumerProps.setProperty("queued.max.message.chunks", "1");
		consumerProps.putAll(secureProps);
		FlinkKafkaConsumerBase<Tuple2<Long, byte[]>> source = kafkaServer.getConsumer(topic, serSchema, consumerProps);
		DataStreamSource<Tuple2<Long, byte[]>> consuming = env.addSource(source);
		consuming.addSink(new SinkFunction<Tuple2<Long, byte[]>>() {
			private int elCnt = 0;
			@Override
			public void invoke(Tuple2<Long, byte[]> value) throws Exception {
				elCnt++;
				if (value.f0 == -1) {
					if (elCnt == 11) {
						throw new SuccessException();
					} else {
						throw new RuntimeException("There have been " + elCnt + " elements");
					}
				}
				if (elCnt > 10) {
					throw new RuntimeException("More than 10 elements seen: " + elCnt);
				}
			}
		});
		Properties producerProps = new Properties();
		producerProps.setProperty("max.request.size", Integer.toString(1024 * 1024 * 15));
		producerProps.setProperty("retries", "3");
		producerProps.putAll(secureProps);
		producerProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerConnectionStrings);
		DataStream<Tuple2<Long, byte[]>> stream = env.addSource(new RichSourceFunction<Tuple2<Long, byte[]>>() {
			private boolean running;
			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				running = true;
			}
			@Override
			public void run(SourceContext<Tuple2<Long, byte[]>> ctx) throws Exception {
				Random rnd = new Random();
				long cnt = 0;
				int sevenMb = 1024 * 1024 * 7;
				while (running) {
					byte[] wl = new byte[sevenMb + rnd.nextInt(sevenMb)];
					ctx.collect(new Tuple2<>(cnt++, wl));
					Thread.sleep(100);
					if (cnt == 10) {
						ctx.collect(new Tuple2<>(-1L, new byte[]{1}));
						break;
					}
				}
			}
			@Override
			public void cancel() {
				running = false;
			}
		});
		kafkaServer.produceIntoKafka(stream, topic, new KeyedSerializationSchemaWrapper<>(serSchema), producerProps, null);
		tryExecute(env, "big topology test");
		deleteTestTopic(topic);
	}
public void runBigRecordTestTopology ( ) throws Exception { final String topic = "bigRecordTestTopic" ; final int parallelism = 1 ; createTestTopic ( topic , parallelism , 1 ) ; final TypeInformation < Tuple2 < Long , byte [ ] > > longBytesInfo = TypeInfoParser . parse ( "Tuple2<Long, byte[]>" ) ; final TypeInformationSerializationSchema < Tuple2 < Long , byte [ ] > > serSchema = new TypeInformationSerializationSchema < > ( longBytesInfo , new ExecutionConfig ( ) ) ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; env . enableCheckpointing ( 100 ) ; env . setParallelism ( parallelism ) ; Properties consumerProps = new Properties ( ) ; consumerProps . putAll ( standardProps ) ; consumerProps . setProperty ( "fetch.message.max.bytes" , Integer . toString ( 1024 * 1024 * 14 ) ) ; consumerProps . setProperty ( "max.partition.fetch.bytes" , Integer . toString ( 1024 * 1024 * 14 ) ) ; consumerProps . setProperty ( "queued.max.message.chunks" , "1" ) ; consumerProps . putAll ( secureProps ) ; FlinkKafkaConsumerBase < Tuple2 < Long , byte [ ] > > source = kafkaServer . getConsumer ( topic , serSchema , consumerProps ) ; DataStreamSource < Tuple2 < Long , byte [ ] > > consuming = env . addSource ( source ) ; consuming . addSink ( new SinkFunction < Tuple2 < Long , byte [ ] > > ( ) { private int elCnt = 0 ; @ Override public void invoke ( Tuple2 < Long , byte [ ] > value ) throws Exception { elCnt ++ ; if ( value . f0 == - 1 ) { if ( elCnt == 11 ) { throw new SuccessException ( ) ; } else { throw new RuntimeException ( "There have been " + elCnt + " elements" ) ; } } if ( elCnt > 10 ) { throw new RuntimeException ( "More than 10 elements seen: " + elCnt ) ; } } } ) ; Properties producerProps = new Properties ( ) ; producerProps . setProperty ( "max.request.size" , Integer . toString ( 1024 * 1024 * 15 ) ) ; producerProps . setProperty ( "retries" , "3" ) ; producerProps . putAll ( secureProps ) ; producerProps . setProperty ( ProducerConfig . BOOTSTRAP_SERVERS_CONFIG , brokerConnectionStrings ) ; DataStream < Tuple2 < Long , byte [ ] > > stream = env . addSource ( new RichSourceFunction < Tuple2 < Long , byte [ ] > > ( ) { private boolean running ; @ Override public void open ( Configuration parameters ) throws Exception { super . open ( parameters ) ; running = true ; } @ Override public void run ( SourceContext < Tuple2 < Long , byte [ ] > > ctx ) throws Exception { Random rnd = new Random ( ) ; long cnt = 0 ; int sevenMb = 1024 * 1024 * 7 ; while ( running ) { byte [ ] wl = new byte [ sevenMb + rnd . nextInt ( sevenMb ) ] ; ctx . collect ( new Tuple2 < > ( cnt ++ , wl ) ) ; Thread . sleep ( 100 ) ; if ( cnt == 10 ) { ctx . collect ( new Tuple2 < > ( - 1L , new byte [ ] { 1 } ) ) ; break ; } } } @ Override public void cancel ( ) { running = false ; } } ) ; kafkaServer . produceIntoKafka ( stream , topic , new KeyedSerializationSchemaWrapper < > ( serSchema ) , producerProps , null ) ; tryExecute ( env , "big topology test" ) ; deleteTestTopic ( topic ) ; }
Test Flink's Kafka integration also with very big records (30MB).
Test Flink ' s Kafka integration also with very big records ( 30MB ) .
// otherwise, the kafka mini clusters may run out of heap space // add consuming topology: // for the new fetcher // we should have seen 11 elements now. // add producing topology // signal end